import argparse import os from contextlib import nullcontext from typing import Any, Dict, Optional, Tuple import safetensors.torch import torch from accelerate import init_empty_weights from huggingface_hub import hf_hub_download from diffusers import LTX2VideoTransformer3DModel from diffusers.utils.import_utils import is_accelerate_available CTX = init_empty_weights if is_accelerate_available() else nullcontext LTX_2_0_TRANSFORMER_KEYS_RENAME_DICT = { # Input Patchify Projections "patchify_proj": "proj_in", "audio_patchify_proj": "audio_proj_in", # Modulation Parameters # Handle adaln_single --> time_embed, audioln_single --> audio_time_embed separately as the original keys are # substrings of the other modulation parameters below "av_ca_video_scale_shift_adaln_single": "av_cross_attn_video_scale_shift", "av_ca_a2v_gate_adaln_single": "av_cross_attn_video_a2v_gate", "av_ca_audio_scale_shift_adaln_single": "av_cross_attn_audio_scale_shift", "av_ca_v2a_gate_adaln_single": "av_cross_attn_audio_v2a_gate", # Transformer Blocks # Per-Block Cross Attention Modulatin Parameters "scale_shift_table_a2v_ca_video": "video_a2v_cross_attn_scale_shift_table", "scale_shift_table_a2v_ca_audio": "audio_a2v_cross_attn_scale_shift_table", # Attention QK Norms "q_norm": "norm_q", "k_norm": "norm_k", } def update_state_dict_inplace(state_dict: Dict[str, Any], old_key: str, new_key: str) -> None: state_dict[new_key] = state_dict.pop(old_key) def remove_keys_inplace(key: str, state_dict: Dict[str, Any]) -> None: state_dict.pop(key) def convert_ltx2_transformer_adaln_single(key: str, state_dict: Dict[str, Any]) -> None: # Skip if not a weight, bias if ".weight" not in key and ".bias" not in key: return if key.startswith("adaln_single."): new_key = key.replace("adaln_single.", "time_embed.") param = state_dict.pop(key) state_dict[new_key] = param if key.startswith("audio_adaln_single."): new_key = key.replace("audio_adaln_single.", "audio_time_embed.") param = state_dict.pop(key) state_dict[new_key] = param return LTX_2_0_TRANSFORMER_SPECIAL_KEYS_REMAP = { "video_embeddings_connector": remove_keys_inplace, "audio_embeddings_connector": remove_keys_inplace, "adaln_single": convert_ltx2_transformer_adaln_single, } def get_ltx2_transformer_config(version: str) -> Tuple[Dict[str, Any], Dict[str, Any], Dict[str, Any]]: if version == "test": # Produces a transformer of the same size as used in test_models_transformer_ltx2.py config = { "model_id": "diffusers-internal-dev/dummy-ltx2", "diffusers_config": { "in_channels": 4, "out_channels": 4, "patch_size": 1, "patch_size_t": 1, "num_attention_heads": 2, "attention_head_dim": 8, "cross_attention_dim": 16, "vae_scale_factors": (8, 32 ,32), "pos_embed_max_pos": 20, "base_height": 2048, "base_width": 2048, "audio_in_channels": 4, "audio_out_channels": 4, "audio_patch_size": 1, "audio_patch_size_t": 1, "audio_num_attention_heads": 2, "audio_attention_head_dim": 4, "audio_cross_attention_dim": 8, "audio_scale_factor": 4, "audio_pos_embed_max_pos": 20, "audio_sampling_rate": 16000, "audio_hop_length": 160, "num_layers": 2, "activation_fn": "gelu-approximate", "qk_norm": "rms_norm_across_heads", "norm_elementwise_affine": False, "norm_eps": 1e-6, "caption_channels": 16, "attention_bias": True, "attention_out_bias": True, "rope_theta": 10000.0, "causal_offset": 1, }, } rename_dict = LTX_2_0_TRANSFORMER_KEYS_RENAME_DICT special_keys_remap = LTX_2_0_TRANSFORMER_SPECIAL_KEYS_REMAP elif version == "2.0": config = { "model_id": "diffusers-internal-dev/new-ltx-model", "diffusers_config": { "in_channels": 128, "out_channels": 128, "patch_size": 1, "patch_size_t": 1, "num_attention_heads": 32, "attention_head_dim": 128, "cross_attention_dim": 4096, "vae_scale_factors": (8, 32 ,32), "pos_embed_max_pos": 20, "base_height": 2048, "base_width": 2048, "audio_in_channels": 128, "audio_out_channels": 128, "audio_patch_size": 1, "audio_patch_size_t": 1, "audio_num_attention_heads": 32, "audio_attention_head_dim": 64, "audio_cross_attention_dim": 2048, "audio_scale_factor": 4, "audio_pos_embed_max_pos": 20, "audio_sampling_rate": 16000, "audio_hop_length": 160, "num_layers": 48, "activation_fn": "gelu-approximate", "qk_norm": "rms_norm_across_heads", "norm_elementwise_affine": False, "norm_eps": 1e-6, "caption_channels": 3840, "attention_bias": True, "attention_out_bias": True, "rope_theta": 10000.0, "causal_offset": 1, }, } rename_dict = LTX_2_0_TRANSFORMER_KEYS_RENAME_DICT special_keys_remap = LTX_2_0_TRANSFORMER_SPECIAL_KEYS_REMAP return config, rename_dict, special_keys_remap def convert_ltx2_transformer(original_state_dict: Dict[str, Any], version: str) -> Dict[str, Any]: config, rename_dict, special_keys_remap = get_ltx2_transformer_config(version) diffusers_config = config["diffusers_config"] with init_empty_weights(): transformer = LTX2VideoTransformer3DModel.from_config(diffusers_config) # Handle official code --> diffusers key remapping via the remap dict for key in list(original_state_dict.keys()): new_key = key[:] for replace_key, rename_key in rename_dict.items(): new_key = new_key.replace(replace_key, rename_key) update_state_dict_inplace(original_state_dict, key, new_key) # Handle any special logic which can't be expressed by a simple 1:1 remapping with the handlers in # special_keys_remap for key in list(original_state_dict.keys()): for special_key, handler_fn_inplace in special_keys_remap.items(): if special_key not in key: continue handler_fn_inplace(key, original_state_dict) transformer.load_state_dict(original_state_dict, strict=True, assign=True) return transformer def load_original_checkpoint(args, filename: Optional[str]) -> Dict[str, Any]: if args.original_state_dict_repo_id is not None: ckpt_path = hf_hub_download(repo_id=args.original_state_dict_repo_id, filename=filename) elif args.checkpoint_path is not None: ckpt_path = args.checkpoint_path else: raise ValueError("Please provide either `original_state_dict_repo_id` or a local `checkpoint_path`") original_state_dict = safetensors.torch.load_file(ckpt_path) return original_state_dict def load_hub_or_local_checkpoint(repo_id: Optional[str] = None, filename: Optional[str] = None) -> Dict[str, Any]: if repo_id is None and filename is None: raise ValueError("Please supply at least one of `repo_id` or `filename`") if repo_id is not None: if filename is None: raise ValueError("If repo_id is specified, filename must also be specified.") ckpt_path = hf_hub_download(repo_id=repo_id, filename=filename) else: ckpt_path = filename _, ext = os.path.splitext(ckpt_path) if ext in [".safetensors", ".sft"]: state_dict = safetensors.torch.load_file(ckpt_path) else: state_dict = torch.load(ckpt_path, map_location="cpu") return state_dict def get_model_state_dict_from_combined_ckpt(combined_ckpt: Dict[str, Any], prefix: str) -> Dict[str, Any]: # Ensure that the key prefix ends with a dot (.) if not prefix.endswith("."): prefix = prefix + "." model_state_dict = {} for param_name, param in combined_ckpt.items(): if param_name.startswith(prefix): model_state_dict[param_name.replace(prefix, "")] = param return model_state_dict def get_args(): parser = argparse.ArgumentParser() parser.add_argument( "--original_state_dict_repo_id", default="diffusers-internal-dev/new-ltx-model", type=str, help="HF Hub repo id with LTX 2.0 checkpoint", ) parser.add_argument( "--checkpoint_path", default=None, type=str, help="Local checkpoint path for LTX 2.0. Will be used if `original_state_dict_repo_id` is not specified.", ) parser.add_argument( "--version", type=str, default="2.0", choices=["test", "2.0"], help="Version of the LTX 2.0 model", ) parser.add_argument( "--combined_filename", default="ltx-av-step-1932500-interleaved-new-vae.safetensors", type=str, help="Filename for combined checkpoint with all LTX 2.0 models (VAE, DiT, etc.)", ) parser.add_argument("--vae_prefix", default="vae.", type=str) parser.add_argument("--audio_vae_prefix", default="audio_vae.", type=str) parser.add_argument("--dit_prefix", default="model.diffusion_model.", type=str) parser.add_argument("--vocoder_prefix", default="vocoder.", type=str) parser.add_argument("--vae_filename", default=None, type=str, help="VAE filename; overrides combined ckpt if set") parser.add_argument( "--audio_vae_filename", default=None, type=str, help="Audio VAE filename; overrides combined ckpt if set" ) parser.add_argument("--dit_filename", default=None, type=str, help="DiT filename; overrides combined ckpt if set") parser.add_argument( "--vocoder_filename", default=None, type=str, help="Vocoder filename; overrides combined ckpt if set" ) parser.add_argument("--vae", action="store_true", help="Whether to convert the video VAE model") parser.add_argument("--audio_vae", action="store_true", help="Whether to convert the audio VAE model") parser.add_argument("--dit", action="store_true", help="Whether to convert the DiT model") parser.add_argument("--vocoder", action="store_true", help="Whether to convert the vocoder model") parser.add_argument( "--full_pipeline", action="store_true", help="Whether to save the pipeline. This will attempt to convert all models (e.g. vae, dit, etc.)", ) parser.add_argument("--vae_dtype", type=str, default="bf16", choices=["fp32", "fp16", "bf16"]) parser.add_argument("--audio_vae_dtype", type=str, default="bf16", choices=["fp32", "fp16", "bf16"]) parser.add_argument("--dit_dtype", type=str, default="bf16", choices=["fp32", "fp16", "bf16"]) parser.add_argument("--vocoder_dtype", type=str, default="bf16", choices=["fp32", "fp16", "bf16"]) parser.add_argument("--output_path", type=str, required=True, help="Path where converted model should be saved") return parser.parse_args() DTYPE_MAPPING = { "fp32": torch.float32, "fp16": torch.float16, "bf16": torch.bfloat16, } VARIANT_MAPPING = { "fp32": None, "fp16": "fp16", "bf16": "bf16", } def main(args): vae_dtype = DTYPE_MAPPING[args.vae_dtype] audio_vae_dtype = DTYPE_MAPPING[args.audio_vae_dtype] dit_dtype = DTYPE_MAPPING[args.dit_dtype] vocoder_dtype = DTYPE_MAPPING[args.vocoder_dtype] combined_ckpt = None load_combined_models = any([args.vae, args.audio_vae, args.dit, args.vocoder, args.full_pipeline]) if args.combined_filename is not None and load_combined_models: combined_ckpt = load_original_checkpoint(args, filename=args.combined_filename) if args.vae or args.full_pipeline: pass if args.audio_vae or args.full_pipeline: pass if args.dit or args.full_pipeline: if args.dit_filename is not None: original_dit_ckpt = load_hub_or_local_checkpoint(filename=args.dit_filename) elif combined_ckpt is not None: original_dit_ckpt = get_model_state_dict_from_combined_ckpt(combined_ckpt, args.dit_prefix) transformer = convert_ltx2_transformer(original_dit_ckpt, version=args.version) if not args.full_pipeline: transformer.to(dit_dtype).save_pretrained(os.path.join(args.output_path, "transformer")) if args.vocoder or args.full_pipeline: pass if args.full_pipeline: pass if __name__ == '__main__': args = get_args() main(args)