enhance automated testing

Signed-off-by: Vladimir Mandic <mandic00@live.com>
This commit is contained in:
Vladimir Mandic
2026-06-03 09:27:34 +02:00
parent c47d2e0dd8
commit ec2e37ee6e
52 changed files with 192 additions and 36 deletions
+3
View File
@@ -93,6 +93,9 @@ Plus continued work on modernization of codebase: UI is now fully TypeScript bas
- large refactor of `lora` native loader
- improve `kanvas` typing
- additional strong typing in core, thanks @awsr
- enhance automated testing
`pnpm test` (uses `--test`) flag runs pipeline init checks
`pnpm compile` (new) runs static python compile and import checks
- **Fixes**
- `hidream-o1` prequant loading
- `gradio` initial hijack
+3 -7
View File
@@ -215,13 +215,9 @@ def start_server(immediate=True, server=None):
uvicorn = None
if args.test:
log.info("Test only")
log.critical('Logging: level=critical')
log.error('Logging: level=error')
log.warning('Logging: level=warning')
log.info('Logging: level=info')
log.debug('Logging: level=debug')
log.trace('Logging: level=trace')
from pipelines.generic_test import test_pipelines
test_pipelines()
log.info("Test only: exiting...")
server.wants_restart = False
else:
uvicorn = server.webui(restart=not immediate)
+4 -1
View File
@@ -172,7 +172,10 @@ def create_sdnq_config(kwargs = None, allow: bool = True, module: str = 'Model',
if quantized_matmul_dtype is None:
quantized_matmul_dtype = "auto" # set for logging
svd = f'{shared.opts.sdnq_use_svd} rank={shared.opts.sdnq_svd_rank} steps={shared.opts.sdnq_svd_steps}' if shared.opts.sdnq_use_svd else f'{shared.opts.sdnq_use_svd}'
log.debug(f'Quantization: module="{module}" type=sdnq mode=pre dtype={weights_dtype} svd={svd} hadamard={shared.opts.sdnq_use_hadamard} dynamic={shared.opts.sdnq_use_dynamic_quantization} group={shared.opts.sdnq_group_size} hadamard_group={shared.opts.sdnq_hadamard_group_size} loss={shared.opts.sdnq_dynamic_loss_threshold} matmul_dtype={quantized_matmul_dtype} matmul_quant={shared.opts.sdnq_use_quantized_matmul} matmul_conv={shared.opts.sdnq_use_quantized_matmul_conv} quant_conv={shared.opts.sdnq_quantize_conv_layers} quant_embedding={shared.opts.sdnq_quantize_embedding_layers} fp32={shared.opts.sdnq_dequantize_fp32} device={quantization_device} return={return_device} use_gpu={shared.opts.sdnq_quantize_with_gpu} map={shared.opts.device_map} offload={shared.opts.diffusers_offload_mode} non_blocking={shared.opts.diffusers_offload_nonblocking} skip_modules={modules_to_not_convert} dict={modules_dtype_dict}')
hadamard = f'{shared.opts.sdnq_use_hadamard} group={shared.opts.sdnq_hadamard_group_size}' if shared.opts.sdnq_use_hadamard else f'{shared.opts.sdnq_use_hadamard}'
log.debug(f'Quantization: module="{module}" type=sdnq mode=pre dtype={weights_dtype} svd={svd} hadamard={hadamard} dynamic={shared.opts.sdnq_use_dynamic_quantization} group={shared.opts.sdnq_group_size} loss={shared.opts.sdnq_dynamic_loss_threshold} matmul_dtype={quantized_matmul_dtype} matmul_quant={shared.opts.sdnq_use_quantized_matmul} matmul_conv={shared.opts.sdnq_use_quantized_matmul_conv} quant_conv={shared.opts.sdnq_quantize_conv_layers} quant_embed={shared.opts.sdnq_quantize_embedding_layers} fp32={shared.opts.sdnq_dequantize_fp32} device={quantization_device} return={return_device} gpu={shared.opts.sdnq_quantize_with_gpu} map={shared.opts.device_map}')
if len(modules_to_not_convert) > 0 or modules_dtype_dict:
log.debug(f'Quantization: module={module} type=sdnq skip_modules={modules_to_not_convert} modules_dtype_dict={modules_dtype_dict}')
if kwargs is None:
return sdnq_config
else:
+2 -2
View File
@@ -29,10 +29,10 @@ class CheckpointInfo:
def __init__(self, filename: str, name: str | None = None, sha: str | None = None, subfolder: str | None = None, model_type: str = 'checkpoint', folder: str | None = None):
self.name = name
self.hash = sha
self.filename = filename
self.filename = filename if filename is not None else 'none'
self.type = ''
self.subfolder = subfolder
relname = filename
relname = self.filename
app_path = os.path.abspath(paths.script_path)
def rel(fn: str, path: str):
+3
View File
@@ -1504,6 +1504,9 @@ def hf_auth_check(checkpoint_info: CheckpointInfo, force:bool=False):
except Exception:
pass
repo_id = path_to_repo(checkpoint_info)
if repo_id is None or '/' not in repo_id:
# log.warning(f'Auth: repo="{repo_id}" invalid repo id')
return False
try:
login = modelloader.hf_login()
return hf.auth_check(repo_id)
+20
View File
@@ -0,0 +1,20 @@
import os
import importlib
from installer import log
def test_pipelines():
from modules.sd_checkpoint import CheckpointInfo
log.info('Test pipelines...')
pipelines = os.listdir("pipelines")
pipelines.sort()
for filename in pipelines:
if filename.startswith("model_") and filename.endswith(".py"):
module_name = filename[:-3]
module = importlib.import_module("pipelines." + module_name)
for attr in dir(module):
if attr.startswith("load_"):
load_func = getattr(module, attr)
log.debug(f"Test: {module_name}.{attr}()")
ckpt = CheckpointInfo(filename = 'none')
load_func(ckpt)
+2
View File
@@ -51,6 +51,8 @@ def load_local_file(local_file, cls_name, quant_type): # t5-only
def load_text_encoder(repo_id, cls_name, load_config=None, subfolder="text_encoder", allow_quant=True, allow_shared=True, variant=None, dtype=None, modules_to_not_convert=None, modules_dtype_dict=None, **kwargs):
if shared.state.interrupted:
return None
if repo_id is None or repo_id.lower() == 'none':
return None
text_encoder = None
allow_shared = allow_shared and shared.opts.te_shared_te
if load_config is None:
+2
View File
@@ -15,6 +15,8 @@ def load_transformer(repo_id, cls_name, load_config=None, subfolder="transformer
to :func:`pipelines.native_transformer.load`. Without a spec, a single-file
override falls back to ``from_single_file``.
"""
if repo_id is None or repo_id.lower() == 'none':
return None
if shared.state.interrupted:
return None
transformer = None
+5 -2
View File
@@ -16,7 +16,7 @@ def _import_from_file(module_name, file_path):
return mod
def load_transformer_components(repo_id, diffusers_load_config, adapter_cls):
def init_transformer_component(repo_id, diffusers_load_config, adapter_cls):
"""Load (transformer, llm_adapter_or_none).
If the UNET dropdown points at a valid safetensors, route through
@@ -57,6 +57,9 @@ def load_anima(checkpoint_info, diffusers_load_config=None):
load_args, _quant_args = model_quant.get_dit_args(diffusers_load_config, allow_quant=False)
log.debug(f'Load model: type=Anima repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={load_args}')
if repo_id is None or repo_id.lower() == 'none':
return None
# load-or-download custom pipeline modules from repo
if os.path.exists(os.path.join(repo_id, 'pipeline.py')):
pipeline_file = os.path.join(repo_id, 'pipeline.py')
@@ -91,7 +94,7 @@ def load_anima(checkpoint_info, diffusers_load_config=None):
# UNET dropdown (shared.opts.sd_unet) may redirect the transformer to a
# community file that bundles both the transformer and the llm_adapter.
transformer, llm_adapter = load_transformer_components(repo_id, diffusers_load_config, AnimaLLMAdapter)
transformer, llm_adapter = init_transformer_component(repo_id, diffusers_load_config, AnimaLLMAdapter)
if transformer is None:
return None
text_encoder = generic.load_text_encoder(
+2
View File
@@ -16,6 +16,8 @@ def load_auraflow(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=diffusers.AuraFlowTransformer2DModel, load_config=diffusers_load_config)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.UMT5EncoderModel, load_config=diffusers_load_config, allow_shared=False) # auraflow uses EleutherAI/pile-t5-xl
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.AuraFlowPipeline.from_pretrained(
repo_id,
+6
View File
@@ -45,6 +45,9 @@ def load_bria(checkpoint_info, diffusers_load_config=None):
cls = diffusers.BriaFiboPipeline
diffusers.pipelines.auto_pipeline.AUTO_TEXT2IMAGE_PIPELINES_MAPPING['bria-fibo'] = cls
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = cls.from_pretrained(
repo_id,
transformer=transformer,
@@ -72,6 +75,9 @@ def load_bria(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=BriaTransformer2DModel, load_config=diffusers_load_config, native_spec=BRIA_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = BriaPipeline.from_pretrained(
repo_id,
transformer=transformer,
+2
View File
@@ -17,6 +17,8 @@ def load_chroma(checkpoint_info, diffusers_load_config=None):
from pipelines.chroma import CHROMA_SPEC
transformer = generic.load_transformer(repo_id, cls_name=diffusers.ChromaTransformer2DModel, load_config=diffusers_load_config, modules_to_not_convert=["distilled_guidance_layer"], native_spec=CHROMA_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.ChromaPipeline.from_pretrained(
repo_id,
+2
View File
@@ -24,6 +24,8 @@ def load_chrono(checkpoint_info, diffusers_load_config=None):
from pipelines.chrono import CHRONOEDIT_SPEC
transformer = generic.load_transformer(repo_id, cls_name=diffusers.ChronoEditTransformer3DModel, load_config=diffusers_load_config, subfolder="transformer", native_spec=CHRONOEDIT_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.UMT5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder")
if repo_id is None or repo_id.lower() == 'none':
return None
try:
pipe = diffusers.ChronoEditPipeline.from_pretrained(
+4
View File
@@ -17,6 +17,8 @@ def load_cogview3(checkpoint_info, diffusers_load_config=None):
from pipelines.cogview import COGVIEW3_SPEC
transformer = generic.load_transformer(repo_id, cls_name=diffusers.CogView3PlusTransformer2DModel, load_config=diffusers_load_config, subfolder="transformer", native_spec=COGVIEW3_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder")
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.CogView3PlusPipeline.from_pretrained(
repo_id,
@@ -44,6 +46,8 @@ def load_cogview4(checkpoint_info, diffusers_load_config=None):
from pipelines.cogview import COGVIEW4_SPEC
transformer = generic.load_transformer(repo_id, cls_name=diffusers.CogView4Transformer2DModel, load_config=diffusers_load_config, subfolder="transformer", native_spec=COGVIEW4_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.GlmModel, load_config=diffusers_load_config, subfolder="text_encoder", allow_quant=True)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.CogView4Pipeline.from_pretrained(
repo_id,
+3
View File
@@ -19,6 +19,9 @@ def load_cosmos_t2i(checkpoint_info, diffusers_load_config=None):
text_encoder = generic.load_text_encoder(repo_te, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder", allow_shared=False) # cosmos does use standard t5
safety_checker = Fake_safety_checker()
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.Cosmos2TextToImagePipeline.from_pretrained(
repo_id,
transformer=transformer,
+5
View File
@@ -29,6 +29,8 @@ def load_ernie_image(checkpoint_info, diffusers_load_config=None):
if not shared.opts.model_ernie_enable_pe:
load_args['pe'] = None
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.ErnieImagePipeline.from_pretrained(
repo_id,
@@ -47,6 +49,9 @@ def load_ernie_image(checkpoint_info, diffusers_load_config=None):
diffusers.pipelines.auto_pipeline.AUTO_IMAGE2IMAGE_PIPELINES_MAPPING["ernieimage"] = ErnieImageImg2ImgPipeline
diffusers.pipelines.auto_pipeline.AUTO_INPAINT_PIPELINES_MAPPING["ernieimage"] = ErnieImageInpaintPipeline
if repo_id is None or repo_id.lower() == 'none':
return None
generic.load_vae_override(pipe, diffusers_load_config)
del transformer
+2
View File
@@ -18,6 +18,8 @@ def load_flex(checkpoint_info, diffusers_load_config=None):
text_encoder_2 = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder_2")
from pipelines.flex2 import Flex2Pipeline
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = Flex2Pipeline.from_pretrained(
repo_id,
transformer=transformer,
+2
View File
@@ -22,6 +22,8 @@ def load_flite(checkpoint_info, diffusers_load_config=None):
dit_model = generic.load_transformer(repo_id, cls_name=f_lite.DiT, load_config=diffusers_load_config, subfolder="dit_model", native_spec=f_lite.FLITE_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder")
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = f_lite.FLitePipeline.from_pretrained(
"Freepik/F-Lite", # pr only exists on main repo
revision="refs/pr/8",
+2
View File
@@ -52,6 +52,8 @@ def load_flux(checkpoint_info, diffusers_load_config=None):
if text_encoder_2 is None:
text_encoder_2 = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = cls_name.from_pretrained(
repo_id,
transformer=transformer,
+2
View File
@@ -16,6 +16,8 @@ def load_flux2(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=diffusers.Flux2Transformer2DModel, load_config=diffusers_load_config)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Mistral3ForConditionalGeneration, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.Flux2Pipeline.from_pretrained(
repo_id,
+4 -5
View File
@@ -14,12 +14,11 @@ def load_flux2_klein(checkpoint_info, diffusers_load_config=None):
load_args, _quant_args = model_quant.get_dit_args(diffusers_load_config, allow_quant=False)
log.debug(f'Load model: type=Flux2Klein repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={load_args}')
# Load transformer - Klein uses Flux2Transformer2DModel (same class as Flux2, different size)
from pipelines.flux2_klein import FLUX2_KLEIN_SPEC
from pipelines.flux2_klein import FLUX2_KLEIN_SPEC # Klein uses Flux2Transformer2DModel (same class as Flux2, different size)
transformer = generic.load_transformer(repo_id, cls_name=diffusers.Flux2Transformer2DModel, load_config=diffusers_load_config, native_spec=FLUX2_KLEIN_SPEC)
# Load text encoder - Klein uses Qwen3 (4B for Klein-4B, 8B for Klein-9B)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Qwen3ForCausalLM, load_config=diffusers_load_config)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Qwen3ForCausalLM, load_config=diffusers_load_config) # Klein uses Qwen3 (4B for Klein-4B, 8B for Klein-9B)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.Flux2KleinPipeline.from_pretrained(
repo_id,
+2
View File
@@ -120,6 +120,8 @@ def load_glm_image(checkpoint_info, diffusers_load_config=None):
load_config=diffusers_load_config,
allow_shared=False
)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.GlmImagePipeline.from_pretrained(
repo_id,
+3 -1
View File
@@ -164,9 +164,11 @@ class GoogleNanoBananaPipeline():
return image
def load_nanobanana(checkpoint_info, diffusers_load_config): # pylint: disable=unused-argument
def load_nanobanana(checkpoint_info, diffusers_load_config=None): # pylint: disable=unused-argument
from modules import sd_models
repo_id = sd_models.path_to_repo(checkpoint_info)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = GoogleNanoBananaPipeline(model_name = repo_id)
return pipe
+6 -2
View File
@@ -5,7 +5,7 @@ from modules.logger import log
from pipelines import generic
def load_llama(diffusers_load_config=None):
def init_llama(diffusers_load_config=None):
if diffusers_load_config is None:
diffusers_load_config = {}
load_args, quant_args = model_quant.get_dit_args(diffusers_load_config, module='TE', device_map=True)
@@ -43,6 +43,8 @@ def load_hidream_o1(checkpoint_info, diffusers_load_config=None):
load_args, quant_args = model_quant.get_dit_args(diffusers_load_config, module='Model', device_map=True, allow_quant=True)
log.debug(f'Load model: type=HiDreamO1 repo="{repo_id}" offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} quant="{model_quant.get_quant_type(quant_args)}" args={load_args}')
if repo_id is None or repo_id.lower() == 'none':
return None
o1_load_config = diffusers_load_config.copy()
o1_load_config['trust_remote_code'] = True
@@ -99,13 +101,14 @@ def load_hidream(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=diffusers.HiDreamImageTransformer2DModel, load_config=diffusers_load_config, subfolder="transformer")
text_encoder_3 = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder_3")
text_encoder_4, tokenizer_4 = load_llama(diffusers_load_config)
if shared.opts.teacache_enabled:
from modules import teacache
log.debug(f'Transformers cache: type=teacache patch=forward cls={diffusers.HiDreamImageTransformer2DModel.__name__}')
diffusers.HiDreamImageTransformer2DModel.forward = teacache.teacache_hidream_forward # patch must be done before transformer is loaded
if repo_id is None or repo_id.lower() == 'none':
return None
if 'I1' in repo_id:
cls = diffusers.HiDreamImagePipeline
elif 'E1' in repo_id:
@@ -122,6 +125,7 @@ def load_hidream(checkpoint_info, diffusers_load_config=None):
log.error(f'Load model: type=HiDream model="{checkpoint_info.name}" repo="{repo_id}" not recognized')
return False
text_encoder_4, tokenizer_4 = init_llama(diffusers_load_config)
pipe = cls.from_pretrained(
repo_id,
transformer=transformer,
+2
View File
@@ -24,6 +24,8 @@ def load_hunyuandit(checkpoint_info, diffusers_load_config=None):
repo_te = 'Tencent-Hunyuan/HunyuanDiT-v1.2-Diffusers' if 'HunyuanDiT-v1' in repo_id else repo_id
text_encoder_2 = generic.load_text_encoder(repo_te, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder_2", allow_shared=False) # this is not normal t5
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.HunyuanDiTPipeline.from_pretrained(
repo_id,
transformer=transformer,
+4
View File
@@ -20,6 +20,8 @@ def load_hyimage(checkpoint_info, diffusers_load_config=None): # pylint: disable
transformer = generic.load_transformer(repo_id, cls_name=diffusers.HunyuanImageTransformer2DModel, load_config=diffusers_load_config, subfolder="transformer", native_spec=HUNYUANIMAGE_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Qwen2_5_VLForConditionalGeneration, load_config=diffusers_load_config, subfolder="text_encoder")
text_encoder_2 = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder_2", allow_shared=False)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.HunyuanImagePipeline.from_pretrained(
repo_id,
@@ -56,6 +58,8 @@ def load_hyimage3(checkpoint_info, diffusers_load_config=None): # pylint: disabl
allow_quant = False
load_args, quant_args = model_quant.get_dit_args(diffusers_load_config, module='Model', device_map=True, allow_quant=allow_quant)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = transformers.AutoModelForCausalLM.from_pretrained(
repo_id,
cache_dir=shared.opts.diffusers_dir,
+2
View File
@@ -26,6 +26,8 @@ def load_joy(checkpoint_info, diffusers_load_config=None):
cls_name=transformers.Qwen3VLForConditionalGeneration,
load_config=diffusers_load_config,
)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.JoyImageEditPipeline.from_pretrained(
repo_id,
+8
View File
@@ -13,6 +13,8 @@ def load_kandinsky21(checkpoint_info, diffusers_load_config=None):
load_args, _quant_args = model_quant.get_dit_args(diffusers_load_config)
log.debug(f'Load model: type=Kandinsky21 repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={load_args}')
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.KandinskyCombinedPipeline.from_pretrained(
repo_id,
cache_dir=shared.opts.diffusers_dir,
@@ -31,6 +33,8 @@ def load_kandinsky22(checkpoint_info, diffusers_load_config=None):
load_args, _quant_args = model_quant.get_dit_args(diffusers_load_config)
log.debug(f'Load model: type=Kandinsky22 repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={load_args}')
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.KandinskyV22CombinedPipeline.from_pretrained(
repo_id,
cache_dir=shared.opts.diffusers_dir,
@@ -54,6 +58,8 @@ def load_kandinsky3(checkpoint_info, diffusers_load_config=None):
unet = generic.load_transformer(repo_id, cls_name=diffusers.Kandinsky3UNet, load_config=diffusers_load_config, subfolder="unet", variant="fp16", native_spec=KANDINSKY3_UNET_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder", variant="fp16", allow_shared=False)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.Kandinsky3Pipeline.from_pretrained(
repo_id,
unet=unet,
@@ -93,6 +99,8 @@ def load_kandinsky5(checkpoint_info, diffusers_load_config=None):
else:
cls = diffusers.Kandinsky5T2IPipeline
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = cls.from_pretrained(
repo_id,
transformer=transformer,
+2
View File
@@ -15,6 +15,8 @@ def load_kolors(checkpoint_info, diffusers_load_config=None):
diffusers_load_config['torch_dtype'] = torch.float16
log.debug(f'Load model: type=Kolors repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={diffusers_load_config}')
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.KolorsPipeline.from_pretrained(
repo_id,
cache_dir = shared.opts.diffusers_dir,
+2
View File
@@ -18,6 +18,8 @@ def load_lens(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=lens.LensTransformer2DModel, load_config=diffusers_load_config, native_spec=lens.LENS_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=lens.LensGptOssEncoder, load_config=diffusers_load_config, allow_quant=False)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = lens.LensPipeline.from_pretrained(
repo_id,
transformer=transformer,
+4
View File
@@ -17,6 +17,10 @@ def load_longcat(checkpoint_info, diffusers_load_config=None):
from pipelines.longcat import LONGCAT_SPEC
transformer = generic.load_transformer(repo_id, cls_name=diffusers.LongCatImageTransformer2DModel, load_config=diffusers_load_config, native_spec=LONGCAT_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Qwen2_5_VLForConditionalGeneration, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
text_processor = transformers.Qwen2VLProcessor.from_pretrained(repo_id, subfolder='tokenizer', cache_dir=shared.opts.hfcache_dir)
if 'edit' in repo_id.lower():
+11 -1
View File
@@ -13,8 +13,12 @@ def load_lumina(checkpoint_info, diffusers_load_config=None):
load_config, _quant_config = model_quant.get_dit_args(diffusers_load_config, allow_quant=False)
log.debug(f'Load model: type=LuminaSFT repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={diffusers_load_config}')
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.LuminaText2ImgPipeline.from_pretrained(
'Alpha-VLLM/Lumina-Next-SFT-diffusers',
repo_id,
cache_dir = shared.opts.diffusers_dir,
**load_config,
)
@@ -41,6 +45,9 @@ def load_lumina2(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=diffusers.Lumina2Transformer2DModel, load_config=diffusers_load_config)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Gemma2Model, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
load_config, _quant_args = model_quant.get_dit_args(diffusers_load_config, allow_quant=False)
pipe = diffusers.Lumina2Pipeline.from_pretrained(
repo_id,
@@ -73,6 +80,9 @@ def load_lumina_dimoo(checkpoint_info, diffusers_load_config=None):
diffusers.pipelines.auto_pipeline.AUTO_TEXT2IMAGE_PIPELINES_MAPPING["luminadimoo"] = LuminaDiMOOTextPipeline
diffusers.pipelines.auto_pipeline.AUTO_IMAGE2IMAGE_PIPELINES_MAPPING["luminadimoo"] = LuminaDiMOOImagePipeline
if repo_id is None or repo_id.lower() == 'none':
return None
# Force slow tokenizer path for Lumina-DiMOO to avoid fast-tokenizer conversion failures.
tokenizer = transformers.AutoTokenizer.from_pretrained(
repo_id,
+2
View File
@@ -21,6 +21,8 @@ def load_meissonic(checkpoint_info, diffusers_load_config=None):
diffusers_load_config['trust_remote_code'] = True
log.debug(f'Load model: type=Meissonic repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={diffusers_load_config}')
if repo_id is None or repo_id.lower() == 'none':
return None
model = TransformerMeissonic.from_pretrained(
repo_id,
subfolder="transformer",
+2
View File
@@ -10,6 +10,8 @@ def load_nextstep(checkpoint_info, diffusers_load_config=None): # pylint: disabl
repo_id = sd_models.path_to_repo(checkpoint_info)
sd_models.hf_auth_check(checkpoint_info)
if repo_id is None or repo_id.lower() == 'none':
return None
log.error(f'Load model: type=NextStep model="{checkpoint_info.name}" repo="{repo_id}" not supported')
"""
+2
View File
@@ -26,6 +26,8 @@ def load_nucleus(checkpoint_info, diffusers_load_config=None):
cls_name=transformers.Qwen3VLForConditionalGeneration,
load_config=diffusers_load_config,
)
if repo_id is None or repo_id.lower() == 'none':
return None
processor = transformers.Qwen3VLProcessor.from_pretrained(
repo_id,
subfolder='processor',
+4
View File
@@ -11,6 +11,8 @@ def load_omnigen(checkpoint_info, diffusers_load_config=None): # pylint: disable
load_config, quant_config = model_quant.get_dit_args(diffusers_load_config, module='Model')
log.debug(f'Load model: type=OmniGen repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={diffusers_load_config}')
if repo_id is None or repo_id.lower() == 'none':
return None
transformer = diffusers.OmniGenTransformer2DModel.from_pretrained(
repo_id,
subfolder="transformer",
@@ -47,6 +49,8 @@ def load_omnigen2(checkpoint_info, diffusers_load_config=None): # pylint: disabl
load_config, quant_config = model_quant.get_dit_args(diffusers_load_config, module='Model')
log.debug(f'Load model: type=OmniGen2 repo="{repo_id}" config={diffusers_load_config} offload={shared.opts.diffusers_offload_mode} dtype={devices.dtype} args={diffusers_load_config}')
if repo_id is None or repo_id.lower() == 'none':
return None
transformer = OmniGen2Transformer2DModel.from_pretrained(
repo_id,
subfolder="transformer",
+3
View File
@@ -18,6 +18,9 @@ def load_ovis(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=diffusers.OvisImageTransformer2DModel, load_config=diffusers_load_config, native_spec=OVIS_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Qwen3Model, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.OvisImagePipeline.from_pretrained(
repo_id,
cache_dir=shared.opts.diffusers_dir,
+4 -2
View File
@@ -16,9 +16,9 @@ def load_pixart(checkpoint_info, diffusers_load_config=None):
repo_id_pipe = repo_id
# PixArt-alpha/PixArt-Sigma-XL-2-2K-MS only holds transformer
if not file_exists(repo_id_tenc, "text_encoder/config.json"):
if not file_exists(repo_id_tenc, "text_encoder/config.json") and repo_id != 'none':
repo_id_tenc = "PixArt-alpha/PixArt-Sigma-XL-2-1024-MS"
if not file_exists(repo_id_pipe, "model_index.json"):
if not file_exists(repo_id_pipe, "model_index.json") and repo_id != 'none':
repo_id_pipe = "PixArt-alpha/PixArt-Sigma-XL-2-1024-MS"
load_args, _quant_args = model_quant.get_dit_args(diffusers_load_config, allow_quant=False)
@@ -28,6 +28,8 @@ def load_pixart(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=diffusers.PixArtTransformer2DModel, load_config=diffusers_load_config, native_spec=PIXART_SPEC)
text_encoder = generic.load_text_encoder(repo_id_tenc, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.PixArtSigmaPipeline.from_pretrained(
repo_id_pipe,
transformer=transformer,
+2
View File
@@ -18,6 +18,8 @@ def load_prx(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=diffusers.PRXTransformer2DModel, load_config=diffusers_load_config, native_spec=PRX_SPEC)
text_encoder = generic.load_text_encoder(repo_id, cls_name=T5GemmaEncoder, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.PRXPipeline.from_pretrained(
repo_id,
transformer=transformer,
+2
View File
@@ -73,6 +73,8 @@ def load_qwen(checkpoint_info, diffusers_load_config=None):
load_config=diffusers_load_config
)
if repo_id is None or repo_id.lower() == 'none':
return None
repo_id, repo_subfolder = qwen.check_qwen_pruning(repo_id, repo_subfolder)
if repo_subfolder is not None and repo_subfolder.startswith('nunchaku'):
repo_subfolder = None
+5 -2
View File
@@ -5,7 +5,7 @@ from modules import shared, sd_models, sd_hijack_te, devices, model_quant
from modules.logger import log
def load_quants(kwargs, repo_id, cache_dir):
def init_quants(kwargs, repo_id, cache_dir):
kwargs_copy = kwargs.copy()
if 'Sana_1600M_1024px' in repo_id and model_quant.check_nunchaku('Model'): # only available model
import nunchaku
@@ -34,6 +34,8 @@ def load_sana(checkpoint_info, kwargs=None):
kwargs.pop('torch_dtype', None)
# set variant since hf repos are a mess
if repo_id is None or repo_id.lower() == 'none':
return None
if not repo_id.endswith('_diffusers'):
repo_id = f'{repo_id}_diffusers'
if 'Sana_1600M' in repo_id:
@@ -47,7 +49,7 @@ def load_sana(checkpoint_info, kwargs=None):
if 'Sana_600M' in repo_id:
kwargs['variant'] = 'fp16'
kwargs = load_quants(kwargs, repo_id, cache_dir=shared.opts.diffusers_dir)
kwargs = init_quants(kwargs, repo_id, cache_dir=shared.opts.diffusers_dir)
log.debug(f'Load model: type=Sana repo="{repo_id}" args={list(kwargs)}')
if devices.dtype == torch.bfloat16 or devices.dtype == torch.float32:
@@ -56,6 +58,7 @@ def load_sana(checkpoint_info, kwargs=None):
cls = diffusers.SanaSprintPipeline
else:
cls = diffusers.SanaPipeline
pipe = cls.from_pretrained(
repo_id,
cache_dir=shared.opts.diffusers_dir,
+2
View File
@@ -22,6 +22,8 @@ def load_sd3(checkpoint_info, diffusers_load_config=None):
else:
text_encoder_3 = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder="text_encoder_3")
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.StableDiffusion3Pipeline.from_pretrained(
repo_id,
transformer=transformer,
+2
View File
@@ -38,6 +38,8 @@ def load_sdxs(checkpoint_info, diffusers_load_config=None):
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Qwen3_5ForConditionalGeneration, load_config=diffusers_load_config, allow_shared=False)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.DiffusionPipeline.from_pretrained(
repo_id,
text_encoder=text_encoder,
+10 -5
View File
@@ -16,7 +16,7 @@ def get_timestep_ratio_conditioning(t, alphas_cumprod):
return ratio
def load_text_encoder(path):
def init_text_encoder(path):
from transformers import CLIPTextConfig, CLIPTextModelWithProjection
from accelerate.utils.modeling import set_module_tensor_to_device
from accelerate import init_empty_weights
@@ -62,7 +62,7 @@ def load_text_encoder(path):
return None
def load_prior(path, config_file="default"):
def init_prior(path, config_file="default"):
from diffusers.models.unets import StableCascadeUNet
prior_text_encoder = None
@@ -78,9 +78,9 @@ def load_prior(path, config_file="default"):
prior_unet = StableCascadeUNet.from_single_file(path, config=config_file, torch_dtype=devices.dtype_unet, cache_dir=shared.opts.diffusers_dir)
if os.path.isfile(os.path.splitext(path)[0] + "_text_encoder.safetensors"): # OneTrainer
prior_text_encoder = load_text_encoder(os.path.splitext(path)[0] + "_text_encoder.safetensors")
prior_text_encoder = init_text_encoder(os.path.splitext(path)[0] + "_text_encoder.safetensors")
elif os.path.isfile(os.path.splitext(path)[0] + "_text_model.safetensors"): # KohyaSS
prior_text_encoder = load_text_encoder(os.path.splitext(path)[0] + "_text_model.safetensors")
prior_text_encoder = init_text_encoder(os.path.splitext(path)[0] + "_text_model.safetensors")
return prior_unet, prior_text_encoder
@@ -97,6 +97,11 @@ def load_cascade_combined(checkpoint_info, diffusers_load_config=None):
if 'cascade' in checkpoint_info.name.lower():
diffusers_load_config["variant"] = 'bf16'
repo_id = sd_models.path_to_repo(checkpoint_info)
sd_models.hf_auth_check(checkpoint_info)
if repo_id is None or repo_id.lower() == 'none':
return None
if shared.opts.sd_unet != "Default" or 'stabilityai' in checkpoint_info.name.lower():
if 'cascade' in checkpoint_info.name and ('lite' in checkpoint_info.name or (checkpoint_info.hash is not None and 'abc818bb0d' in checkpoint_info.hash)):
decoder_folder = 'decoder_lite'
@@ -112,7 +117,7 @@ def load_cascade_combined(checkpoint_info, diffusers_load_config=None):
# log.debug(f'StableCascade {decoder_folder}: scale={decoder.latent_dim_scale}')
prior_text_encoder = None
if shared.opts.sd_unet != "Default":
prior_unet, prior_text_encoder = load_prior(unet_dict[shared.opts.sd_unet])
prior_unet, prior_text_encoder = init_prior(unet_dict[shared.opts.sd_unet])
else:
prior_unet = StableCascadeUNet.from_pretrained("stabilityai/stable-cascade-prior", subfolder=prior_folder, cache_dir=shared.opts.diffusers_dir, **diffusers_load_config)
if prior_text_encoder is not None:
+3 -1
View File
@@ -21,9 +21,11 @@ def load_step1x_edit(checkpoint_info, diffusers_load_config=None):
diffusers.Step1XEditTransformer2DModel = Step1XEditTransformer2DModel
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Qwen2_5_VLForConditionalGeneration, load_config=diffusers_load_config)
processor = transformers.Qwen2_5_VLProcessor.from_pretrained(repo_id, cache_dir=shared.opts.hfcache_dir, subfolder='processor')
from pipelines.step1x import STEP1X_SPEC
transformer = generic.load_transformer(repo_id, cls_name=Step1XEditTransformer2DModel, load_config=diffusers_load_config, native_spec=STEP1X_SPEC)
if repo_id is None or repo_id.lower() == 'none':
return None
processor = transformers.Qwen2_5_VLProcessor.from_pretrained(repo_id, cache_dir=shared.opts.hfcache_dir, subfolder='processor')
pipe = Step1XEditPipeline.from_pretrained(
repo_id,
+3
View File
@@ -20,6 +20,9 @@ def load_ultraflux(checkpoint_info, diffusers_load_config=None):
transformer = generic.load_transformer(repo_id, cls_name=FluxTransformer2DModel, load_config=diffusers_load_config)
text_encoder_2 = generic.load_text_encoder(repo_id, cls_name=transformers.T5EncoderModel, load_config=diffusers_load_config, subfolder='text_encoder_2')
if repo_id is None or repo_id.lower() == 'none':
return None
vae = AutoencoderUltraFluxKL.from_pretrained(
repo_id,
subfolder='vae',
+3
View File
@@ -36,6 +36,9 @@ def load_vibe(checkpoint_info, diffusers_load_config=None):
allow_quant=False,
allow_shared=False,
)
if repo_id is None or repo_id.lower() == 'none':
return None
processor = transformers.Qwen3VLProcessor.from_pretrained(
repo_id,
subfolder='tokenizer',
+4 -2
View File
@@ -5,7 +5,7 @@ from modules.logger import log
from pipelines import generic
def load_text_encoder(repo_id, diffusers_load_config=None):
def init_text_encoder(repo_id, diffusers_load_config=None):
if diffusers_load_config is None:
diffusers_load_config = {}
load_args, quant_args = model_quant.get_dit_args(diffusers_load_config, module='TE', device_map=True)
@@ -50,8 +50,10 @@ def load_wan(checkpoint_info, diffusers_load_config=None):
else:
transformer = generic.load_transformer(repo_id, cls_name=transformer_cls, load_config=diffusers_load_config, subfolder='transformer')
transformer_2 = None
if repo_id is None or repo_id.lower() == 'none':
return None
text_encoder = load_text_encoder(repo_id, diffusers_load_config)
text_encoder = init_text_encoder(repo_id, diffusers_load_config)
load_args, _quant_args = model_quant.get_dit_args(diffusers_load_config, module='Model')
+3
View File
@@ -112,6 +112,9 @@ def load_xomni(checkpoint_info, diffusers_load_config=None):
sd_models.hf_auth_check(checkpoint_info)
pipe = XOmniPipeline()
if repo_id is None or repo_id.lower() == 'none':
return None
pipe.load(repo_id, load_config=diffusers_load_config)
devices.torch_gc(force=True, reason='load')
return pipe
+5 -2
View File
@@ -7,7 +7,7 @@ from modules.logger import log
from pipelines import generic
def load_nunchaku():
def init_nunchaku():
import nunchaku
if not hasattr(nunchaku, 'NunchakuZImageTransformer2DModel'): # not present in older versions of nunchaku
return None
@@ -40,12 +40,15 @@ def load_z_image(checkpoint_info, diffusers_load_config=None):
transformer = None
if model_quant.check_nunchaku('Model'): # only available model
transformer = load_nunchaku()
transformer = init_nunchaku()
if transformer is None:
transformer = generic.load_transformer(repo_id, cls_name=diffusers.ZImageTransformer2DModel, load_config=diffusers_load_config)
text_encoder = generic.load_text_encoder(repo_id, cls_name=transformers.Qwen3ForCausalLM, load_config=diffusers_load_config)
if repo_id is None or repo_id.lower() == 'none':
return None
pipe = diffusers.ZImagePipeline.from_pretrained(
repo_id,
cache_dir=shared.opts.diffusers_dir,
+4
View File
@@ -27,11 +27,15 @@ def load_zetachroma(checkpoint_info, diffusers_load_config=None):
diffusers.ZetaChromaPipeline = zetachroma.ZetaChromaPipeline
sys.modules["zetachroma"] = zetachroma
if repo_id is None or repo_id.lower() == 'none':
return None
text_encoder = generic.load_text_encoder(
TEXT_ENCODER_REPO,
cls_name=transformers.Qwen3ForCausalLM,
load_config=diffusers_load_config,
)
tokenizer = transformers.AutoTokenizer.from_pretrained(
TEXT_ENCODER_REPO,
subfolder="tokenizer",
+1 -1
Submodule wiki updated: d11c0ea0f0...2f47d4cc38