




'''simple docstring'''
import argparse

import huggingface_hub
import k_diffusion as K
import torch

from diffusers import UNetaDConditionModel


A				: List[str]					      =			"""pcuenq/k-upscaler"""
def 							snake_case_    (  a__						:      Any       ,a__						:      Optional[int]       ,*, a__						:      Tuple       ,a__						:      List[str]     ):



    """simple docstring"""




    __lowercase     							=		{
        # norm1
        f'{diffusers_resnet_prefix}.norm1.linear.weight': checkpoint[f'{resnet_prefix}.main.0.mapper.weight'],
        f'{diffusers_resnet_prefix}.norm1.linear.bias': checkpoint[f'{resnet_prefix}.main.0.mapper.bias'],
        # conv1
        f'{diffusers_resnet_prefix}.conv1.weight': checkpoint[f'{resnet_prefix}.main.2.weight'],
        f'{diffusers_resnet_prefix}.conv1.bias': checkpoint[f'{resnet_prefix}.main.2.bias'],
        # norm2
        f'{diffusers_resnet_prefix}.norm2.linear.weight': checkpoint[f'{resnet_prefix}.main.4.mapper.weight'],
        f'{diffusers_resnet_prefix}.norm2.linear.bias': checkpoint[f'{resnet_prefix}.main.4.mapper.bias'],
        # conv2
        f'{diffusers_resnet_prefix}.conv2.weight': checkpoint[f'{resnet_prefix}.main.6.weight'],
        f'{diffusers_resnet_prefix}.conv2.bias': checkpoint[f'{resnet_prefix}.main.6.bias'],
    }

    if resnet.conv_shortcut is not None:
        rv.update(
            {
                f'{diffusers_resnet_prefix}.conv_shortcut.weight': checkpoint[f'{resnet_prefix}.skip.weight'],
            }     )

    return rv
def 							snake_case_    (  a__						:      Dict       ,*, a__						:      List[str]       ,a__						:      Optional[Any]     ):



    """simple docstring"""




    __lowercase							,__lowercase							,__lowercase     							=		checkpoint[f'{attention_prefix}.qkv_proj.weight'].chunk(3       ,dim=0     )
    __lowercase							,__lowercase							,__lowercase     							=		checkpoint[f'{attention_prefix}.qkv_proj.bias'].chunk(3       ,dim=0     )
    __lowercase     							=		{
        # norm
        f'{diffusers_attention_prefix}.norm1.linear.weight': checkpoint[f'{attention_prefix}.norm_in.mapper.weight'],
        f'{diffusers_attention_prefix}.norm1.linear.bias': checkpoint[f'{attention_prefix}.norm_in.mapper.bias'],
        # to_q
        f'{diffusers_attention_prefix}.attn1.to_q.weight': weight_q.squeeze(-1     ).squeeze(-1     ),
        f'{diffusers_attention_prefix}.attn1.to_q.bias': bias_q,
        # to_k
        f'{diffusers_attention_prefix}.attn1.to_k.weight': weight_k.squeeze(-1     ).squeeze(-1     ),
        f'{diffusers_attention_prefix}.attn1.to_k.bias': bias_k,
        # to_v
        f'{diffusers_attention_prefix}.attn1.to_v.weight': weight_v.squeeze(-1     ).squeeze(-1     ),
        f'{diffusers_attention_prefix}.attn1.to_v.bias': bias_v,
        # to_out
        f'{diffusers_attention_prefix}.attn1.to_out.0.weight': checkpoint[f'{attention_prefix}.out_proj.weight']
        .squeeze(-1     )
        .squeeze(-1     ),
        f'{diffusers_attention_prefix}.attn1.to_out.0.bias': checkpoint[f'{attention_prefix}.out_proj.bias'],
    }

    return rv
def 							snake_case_    (  a__						:      List[Any]       ,*, a__						:      str       ,a__						:      List[Any]       ,a__						:      Optional[int]     ):



    """simple docstring"""




    __lowercase							,__lowercase     							=		checkpoint[f'{attention_prefix}.kv_proj.weight'].chunk(2       ,dim=0     )
    __lowercase							,__lowercase     							=		checkpoint[f'{attention_prefix}.kv_proj.bias'].chunk(2       ,dim=0     )

    __lowercase     							=		{
        # norm2 (ada groupnorm)
        f'{diffusers_attention_prefix}.norm{diffusers_attention_index}.linear.weight': checkpoint[
            f'{attention_prefix}.norm_dec.mapper.weight'
        ],
        f'{diffusers_attention_prefix}.norm{diffusers_attention_index}.linear.bias': checkpoint[
            f'{attention_prefix}.norm_dec.mapper.bias'
        ],
        # layernorm on encoder_hidden_state
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.norm_cross.weight': checkpoint[
            f'{attention_prefix}.norm_enc.weight'
        ],
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.norm_cross.bias': checkpoint[
            f'{attention_prefix}.norm_enc.bias'
        ],
        # to_q
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_q.weight': checkpoint[
            f'{attention_prefix}.q_proj.weight'
        ]
        .squeeze(-1     )
        .squeeze(-1     ),
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_q.bias': checkpoint[
            f'{attention_prefix}.q_proj.bias'
        ],
        # to_k
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_k.weight': weight_k.squeeze(-1     ).squeeze(-1     ),
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_k.bias': bias_k,
        # to_v
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_v.weight': weight_v.squeeze(-1     ).squeeze(-1     ),
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_v.bias': bias_v,
        # to_out
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_out.0.weight': checkpoint[
            f'{attention_prefix}.out_proj.weight'
        ]
        .squeeze(-1     )
        .squeeze(-1     ),
        f'{diffusers_attention_prefix}.attn{diffusers_attention_index}.to_out.0.bias': checkpoint[
            f'{attention_prefix}.out_proj.bias'
        ],
    }

    return rv
def 							snake_case_    (  a__						:      List[Any]       ,a__						:      Dict       ,a__						:      Any       ,a__						:      Dict     ):



    """simple docstring"""




    __lowercase     							=		"""inner_model.u_net.u_blocks""" if block_type == """up""" else """inner_model.u_net.d_blocks"""
    __lowercase     							=		f'{block_prefix}.{block_idx}'

    __lowercase     							=		{}

    if not hasattr(a__       ,"""attentions"""     ):
        __lowercase     							=		1  # resnet only
    elif not block.attentions[0].add_self_attention:
        __lowercase     							=		2  # resnet -> cross-attention
    else:
        __lowercase     							=		3  # resnet -> self-attention -> cross-attention)

    for resnet_idx, resnet in enumerate(block.resnets     ):
        # diffusers_resnet_prefix = f"{diffusers_up_block_prefix}.resnets.{resnet_idx}"
        __lowercase     							=		f'{block_type}_blocks.{block_idx}.resnets.{resnet_idx}'
        __lowercase     							=		n * resnet_idx if block_type == """up""" else n * resnet_idx + 1
        __lowercase     							=		f'{block_prefix}.{idx}' if block_type == """up""" else f'{block_prefix}.{idx}'

        diffusers_checkpoint.update(
            resnet_to_diffusers_checkpoint(
                a__       ,a__       ,diffusers_resnet_prefix=a__       ,resnet_prefix=a__     )     )

    if hasattr(a__       ,"""attentions"""     ):
        for attention_idx, attention in enumerate(block.attentions     ):
            __lowercase     							=		f'{block_type}_blocks.{block_idx}.attentions.{attention_idx}'
            __lowercase     							=		n * attention_idx + 1 if block_type == """up""" else n * attention_idx + 2
            __lowercase     							=		f'{block_prefix}.{idx}'
            __lowercase     							=		f'{block_prefix}.{idx }'
            __lowercase     							=		1 if not attention.add_self_attention else 2
            __lowercase     							=		(
                n * attention_idx + cross_attention_index
                if block_type == """up"""
                else n * attention_idx + cross_attention_index + 1
            )
            __lowercase     							=		f'{block_prefix}.{idx }'

            diffusers_checkpoint.update(
                cross_attn_to_diffusers_checkpoint(
                    a__       ,diffusers_attention_prefix=a__       ,diffusers_attention_index=2       ,attention_prefix=a__       ,)     )

            if attention.add_self_attention is True:
                diffusers_checkpoint.update(
                    self_attn_to_diffusers_checkpoint(
                        a__       ,diffusers_attention_prefix=a__       ,attention_prefix=a__       ,)     )

    return diffusers_checkpoint
def 							snake_case_    (  a__						:      str       ,a__						:      Tuple     ):



    """simple docstring"""




    __lowercase     							=		{}

    # pre-processing
    diffusers_checkpoint.update(
        {
            """conv_in.weight""": checkpoint["""inner_model.proj_in.weight"""],
            """conv_in.bias""": checkpoint["""inner_model.proj_in.bias"""],
        }     )

    # timestep and class embedding
    diffusers_checkpoint.update(
        {
            """time_proj.weight""": checkpoint["""inner_model.timestep_embed.weight"""].squeeze(-1     ),
            """time_embedding.linear_1.weight""": checkpoint["""inner_model.mapping.0.weight"""],
            """time_embedding.linear_1.bias""": checkpoint["""inner_model.mapping.0.bias"""],
            """time_embedding.linear_2.weight""": checkpoint["""inner_model.mapping.2.weight"""],
            """time_embedding.linear_2.bias""": checkpoint["""inner_model.mapping.2.bias"""],
            """time_embedding.cond_proj.weight""": checkpoint["""inner_model.mapping_cond.weight"""],
        }     )

    # down_blocks
    for down_block_idx, down_block in enumerate(model.down_blocks     ):
        diffusers_checkpoint.update(block_to_diffusers_checkpoint(a__       ,a__       ,a__       ,"""down"""     )     )

    # up_blocks
    for up_block_idx, up_block in enumerate(model.up_blocks     ):
        diffusers_checkpoint.update(block_to_diffusers_checkpoint(a__       ,a__       ,a__       ,"""up"""     )     )

    # post-processing
    diffusers_checkpoint.update(
        {
            """conv_out.weight""": checkpoint["""inner_model.proj_out.weight"""],
            """conv_out.bias""": checkpoint["""inner_model.proj_out.bias"""],
        }     )

    return diffusers_checkpoint
def 							snake_case_    (  a__						:      List[Any]     ):



    """simple docstring"""




    __lowercase     							=		original_config["""input_channels"""] + original_config["""unet_cond_dim"""]
    __lowercase     							=		original_config["""input_channels"""] + (1 if original_config["""has_variance"""] else 0)

    __lowercase     							=		original_config["""channels"""]

    assert (
        len(set(original_config["""depths"""]     )     ) == 1
    ), "UNet2DConditionModel currently do not support blocks with different number of layers"
    __lowercase     							=		original_config["""depths"""][0]

    __lowercase     							=		original_config["""mapping_cond_dim"""]
    __lowercase     							=		original_config["""cross_cond_dim"""]

    __lowercase     							=		[]
    __lowercase     							=		[]
    for s, c in zip(original_config["""self_attn_depths"""]       ,original_config["""cross_attn_depths"""]     ):
        if s:
            __lowercase     							=		"""self"""
            __lowercase     							=		"""cross""" if c else None
        elif c:
            __lowercase     							=		"""cross"""
            __lowercase     							=		None
        else:
            __lowercase     							=		None
            __lowercase     							=		None
        attna_types.append(a__     )
        attna_types.append(a__     )

    __lowercase     							=		UNetaDConditionModel(
        in_channels=a__       ,out_channels=a__       ,down_block_types=("""KDownBlock2D""", """KCrossAttnDownBlock2D""", """KCrossAttnDownBlock2D""", """KCrossAttnDownBlock2D""")       ,mid_block_type=a__       ,up_block_types=("""KCrossAttnUpBlock2D""", """KCrossAttnUpBlock2D""", """KCrossAttnUpBlock2D""", """KUpBlock2D""")       ,block_out_channels=a__       ,layers_per_block=a__       ,act_fn="""gelu"""       ,norm_num_groups=a__       ,cross_attention_dim=a__       ,attention_head_dim=64       ,time_cond_proj_dim=a__       ,resnet_time_scale_shift="""scale_shift"""       ,time_embedding_type="""fourier"""       ,timestep_post_act="""gelu"""       ,conv_in_kernel=1       ,conv_out_kernel=1       ,)

    return unet
def 							snake_case_    (  a__						:      Optional[Any]     ):



    """simple docstring"""




    __lowercase     							=		torch.device("""cuda""" if torch.cuda.is_available() else """cpu"""     )

    __lowercase     							=		huggingface_hub.hf_hub_download(a__       ,"""config_laion_text_cond_latent_upscaler_2.json"""     )
    __lowercase     							=		huggingface_hub.hf_hub_download(
        a__       ,"""laion_text_cond_latent_upscaler_2_1_00470000_slim.pth"""     )
    print(f'loading original model configuration from {orig_config_path}'     )
    print(f'loading original model checkpoint from {orig_weights_path}'     )

    print("""converting to diffusers unet"""     )
    __lowercase     							=		K.config.load_config(open(a__     )     )["""model"""]
    __lowercase     							=		unet_model_from_original_config(a__     )

    __lowercase     							=		torch.load(a__       ,map_location=a__     )["""model_ema"""]
    __lowercase     							=		unet_to_diffusers_checkpoint(a__       ,a__     )

    model.load_state_dict(a__       ,strict=a__     )
    model.save_pretrained(args.dump_path     )
    print(f'saving converted unet model in {args.dump_path}'     )


if __name__ == "__main__":
  A				: List[Any]					      =			argparse.ArgumentParser()

  parser.add_argument("""--dump_path""", default=None, type=str, required=True, help="""Path to the output model.""")
  A				: int					      =			parser.parse_args()

  main(args)










'''simple docstring'''
import argparse
import json
import os.path
from collections import OrderedDict

import numpy as np
import requests
import torch
from flax.training.checkpoints import restore_checkpoint
from huggingface_hub import hf_hub_download

from transformers import VivitConfig, VivitForVideoClassification, VivitImageProcessor
from transformers.image_utils import PILImageResampling
def 							snake_case_    (  a__						:      Any     ):



    """simple docstring"""




    __lowercase     							=		"""https://storage.googleapis.com/scenic-bucket/vivit/kinetics_400/vivit_base_16x2_unfactorized/checkpoint"""

    with open(a__       ,"""wb"""     ) as f:
        with requests.get(a__       ,stream=a__     ) as req:
            for chunk in req.iter_content(chunk_size=20_48     ):
                f.write(a__     )
def 							snake_case_    (  ):



    """simple docstring"""




    __lowercase     							=		VivitConfig()

    __lowercase     							=		4_00
    __lowercase     							=		"""huggingface/label-files"""
    __lowercase     							=		"""kinetics400-id2label.json"""

    __lowercase     							=		json.load(open(hf_hub_download(a__       ,a__       ,repo_type="""dataset"""     )       ,"""r"""     )     )
    __lowercase     							=		{int(a__     ): v for k, v in idalabel.items()}
    __lowercase     							=		idalabel
    __lowercase     							=		{v: k for k, v in idalabel.items()}
    return config
def 							snake_case_    (  ):



    """simple docstring"""




    __lowercase     							=		hf_hub_download(
        repo_id="""hf-internal-testing/spaghetti-video"""       ,filename="""eating_spaghetti_32_frames.npy"""       ,repo_type="""dataset"""     )
    __lowercase     							=		np.load(a__     )
    return list(a__     )
def 							snake_case_    (  a__						:      np.ndarray     ):



    """simple docstring"""




    if np.ndim(a__     ) == 2:
        return transform_attention_bias(a__     )

    elif np.ndim(a__     ) == 3:
        return transform_attention_kernel(a__     )

    else:
        raise Exception(f'Invalid number of dimesions: {np.ndim(a__     )}'     )
def 							snake_case_    (  a__						:      np.ndarray     ):



    """simple docstring"""




    return current.flatten()
def 							snake_case_    (  a__						:      np.ndarray     ):



    """simple docstring"""




    return np.reshape(a__       ,(current.shape[0], current.shape[1] * current.shape[2])     ).T
def 							snake_case_    (  a__						:      np.ndarray     ):



    """simple docstring"""




    return np.reshape(a__       ,(current.shape[0] * current.shape[1], current.shape[2])     ).T
def 							snake_case_    (  a__						:      Tuple       ,a__						:      Dict     ):



    """simple docstring"""




    __lowercase     							=		state_dict["""optimizer"""]["""target"""]["""Transformer"""][f'encoderblock_{i}']

    __lowercase     							=		f'encoder.layer.{i}.'
    __lowercase     							=		{
        prefix + """intermediate.dense.bias""": state["""MlpBlock_0"""]["""Dense_0"""]["""bias"""],
        prefix + """intermediate.dense.weight""": np.transpose(state["""MlpBlock_0"""]["""Dense_0"""]["""kernel"""]     ),
        prefix + """output.dense.bias""": state["""MlpBlock_0"""]["""Dense_1"""]["""bias"""],
        prefix + """output.dense.weight""": np.transpose(state["""MlpBlock_0"""]["""Dense_1"""]["""kernel"""]     ),
        prefix + """layernorm_before.bias""": state["""LayerNorm_0"""]["""bias"""],
        prefix + """layernorm_before.weight""": state["""LayerNorm_0"""]["""scale"""],
        prefix + """layernorm_after.bias""": state["""LayerNorm_1"""]["""bias"""],
        prefix + """layernorm_after.weight""": state["""LayerNorm_1"""]["""scale"""],
        prefix
        + """attention.attention.query.bias""": transform_attention(
            state["""MultiHeadDotProductAttention_0"""]["""query"""]["""bias"""]     ),
        prefix
        + """attention.attention.query.weight""": transform_attention(
            state["""MultiHeadDotProductAttention_0"""]["""query"""]["""kernel"""]     ),
        prefix
        + """attention.attention.key.bias""": transform_attention(state["""MultiHeadDotProductAttention_0"""]["""key"""]["""bias"""]     ),
        prefix
        + """attention.attention.key.weight""": transform_attention(
            state["""MultiHeadDotProductAttention_0"""]["""key"""]["""kernel"""]     ),
        prefix
        + """attention.attention.value.bias""": transform_attention(
            state["""MultiHeadDotProductAttention_0"""]["""value"""]["""bias"""]     ),
        prefix
        + """attention.attention.value.weight""": transform_attention(
            state["""MultiHeadDotProductAttention_0"""]["""value"""]["""kernel"""]     ),
        prefix + """attention.output.dense.bias""": state["""MultiHeadDotProductAttention_0"""]["""out"""]["""bias"""],
        prefix
        + """attention.output.dense.weight""": transform_attention_output_weight(
            state["""MultiHeadDotProductAttention_0"""]["""out"""]["""kernel"""]     ),
    }

    return new_state
def 							snake_case_    (  a__						:      str     ):



    """simple docstring"""




    return sum([1 if """encoderblock_""" in k else 0 for k in state_dict["""optimizer"""]["""target"""]["""Transformer"""].keys()]     )
def 							snake_case_    (  a__						:      Any       ,a__						:      int=False     ):



    """simple docstring"""




    __lowercase     							=		get_n_layers(a__     )

    __lowercase     							=		OrderedDict()

    __lowercase     							=		state_dict["""optimizer"""]["""target"""]["""Transformer"""]["""encoder_norm"""]["""bias"""]
    __lowercase     							=		state_dict["""optimizer"""]["""target"""]["""Transformer"""]["""encoder_norm"""]["""scale"""]

    __lowercase     							=		np.transpose(
        state_dict["""optimizer"""]["""target"""]["""embedding"""]["""kernel"""]       ,(4, 3, 0, 1, 2)     )
    __lowercase     							=		state_dict["""optimizer"""]["""target"""]["""embedding"""]["""bias"""]

    __lowercase     							=		state_dict["""optimizer"""]["""target"""]["""cls"""]
    __lowercase     							=		state_dict["""optimizer"""]["""target"""]["""Transformer"""]["""posembed_input"""][
        """pos_embedding"""
    ]

    for i in range(a__     ):
        new_state.update(transform_state_encoder_block(a__       ,a__     )     )

    if classification_head:
        __lowercase     							=		{"""vivit.""" + k: v for k, v in new_state.items()}
        __lowercase     							=		np.transpose(state_dict["""optimizer"""]["""target"""]["""output_projection"""]["""kernel"""]     )
        __lowercase     							=		np.transpose(state_dict["""optimizer"""]["""target"""]["""output_projection"""]["""bias"""]     )

    return {k: torch.tensor(a__     ) for k, v in new_state.items()}
def 							snake_case_    (  ):



    """simple docstring"""




    __lowercase     							=		VivitImageProcessor()

    assert extractor.do_resize is True
    assert extractor.size == {"shortest_edge": 2_56}
    assert extractor.do_center_crop is True
    assert extractor.crop_size == {"width": 2_24, "height": 2_24}
    assert extractor.resample == PILImageResampling.BILINEAR

    # here: https://github.com/deepmind/dmvr/blob/master/dmvr/modalities.py
    # one can seen that add_image has default values for normalization_mean and normalization_std set to 0 and 1
    # which effectively means no normalization (and ViViT does not overwrite those when calling this func)
    assert extractor.do_normalize is False
    assert extractor.do_rescale is True
    assert extractor.rescale_factor == 1 / 2_55

    # zero-centering = True in original implementation
    assert extractor.do_zero_centering is True

    return extractor
def 							snake_case_    (  a__						:      str     ):



    """simple docstring"""




    __lowercase     							=		"""checkpoint"""

    if not os.path.exists(a__     ):
        download_checkpoint(a__     )

    __lowercase     							=		restore_checkpoint(a__       ,a__     )
    __lowercase     							=		transform_state(a__       ,classification_head=a__     )

    __lowercase     							=		get_vivit_config()

    assert config.image_size == 2_24
    assert config.num_frames == 32

    __lowercase     							=		VivitForVideoClassification(a__     )
    model.load_state_dict(a__     )
    model.eval()

    __lowercase     							=		get_processor()

    __lowercase     							=		prepare_video()
    __lowercase     							=		extractor(a__       ,return_tensors="""pt"""     )

    __lowercase     							=		model(**a__     )

    __lowercase     							=		torch.Size([1, 4_00]     )
    __lowercase     							=		torch.tensor([-1.0_5_4_3, 2.0_7_6_4, -0.2_1_0_4, 0.4_4_3_9, -0.9_6_5_8]     )

    assert outputs.logits.shape == expected_shape
    assert torch.allclose(outputs.logits[0, :5]       ,a__       ,atol=1e-4     ), outputs.logits[0, :5]

    model.save_pretrained(a__     )
    extractor.save_pretrained(a__     )


if __name__ == "__main__":
  A				: Union[str, Any]					      =			argparse.ArgumentParser()

  parser.add_argument("""--output_model_name""", """-o""", type=str, help="""Output path for the converted HuggingFace model""")

  A				: Optional[Any]					      =			parser.parse_args()
  convert(args.output_model_name)




