





"""simple docstring"""
from logging import getLogger
from typing import Any, Callable, List, Optional, Union

import numpy as np
import PIL
import torch

from ...schedulers import DDPMScheduler
from ..onnx_utils import ORT_TO_NP_TYPE, OnnxRuntimeModel
from ..pipeline_utils import ImagePipelineOutput
from . import StableDiffusionUpscalePipeline


lowerCamelCase_          =	getLogger(__name__)


lowerCamelCase_          =	4
lowerCamelCase_          =	7

lowerCamelCase_          =	{
    "float16": torch.floataa,
    "float32": torch.floataa,
}



def   __lowerCamelCase     (							a_  :	Tuple							)    ->   int:
      if isinstance(a_				,     torch.Tensor							):
            return image
      elif isinstance(a_				,     PIL.Image.Image							):
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						[image]

      if isinstance(image[0]				,     PIL.Image.Image							):
            __SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE     :List[Any]	       =						image[0].size
            __SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE     :Dict	       =						(x - x % 64 for x in (w, h))  # resize to integer multiple of 32

            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						[np.array(i.resize((w, h)							)							)[None, :] for i in image]
            __SCREAMING_SNAKE_CASE     :Dict	       =						np.concatenate(a_				,     axis=0							)
            __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						np.array(a_							).astype(np.floataa							) / 255.0
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						image.transpose(0				,     3				,     1				,     2							)
            __SCREAMING_SNAKE_CASE     :int	       =						2.0 * image - 1.0
            __SCREAMING_SNAKE_CASE     :Tuple	       =						torch.from_numpy(a_							)
      elif isinstance(image[0]				,     torch.Tensor							):
            __SCREAMING_SNAKE_CASE     :str	       =						torch.cat(a_				,     dim=0							)

      return image







class       _SCREAMING_SNAKE_CASE(     A						):




       def __init__(		self	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__ = 3_50	,)    ->   Optional[int]:





             """simple docstring"""






             super().__init__(
                 vae=SCREAMING_SNAKE_CASE__	,text_encoder=SCREAMING_SNAKE_CASE__	,tokenizer=SCREAMING_SNAKE_CASE__	,unet=SCREAMING_SNAKE_CASE__	,low_res_scheduler=SCREAMING_SNAKE_CASE__	,scheduler=SCREAMING_SNAKE_CASE__	,safety_checker=SCREAMING_SNAKE_CASE__	,feature_extractor=SCREAMING_SNAKE_CASE__	,watermarker=SCREAMING_SNAKE_CASE__	,max_noise_level=SCREAMING_SNAKE_CASE__	,)




       def __call__(		self	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__ = 75	,SCREAMING_SNAKE_CASE__ = 9.0	,SCREAMING_SNAKE_CASE__ = 20	,SCREAMING_SNAKE_CASE__ = None	,SCREAMING_SNAKE_CASE__ = 1	,SCREAMING_SNAKE_CASE__ = 0.0	,SCREAMING_SNAKE_CASE__ = None	,SCREAMING_SNAKE_CASE__ = None	,SCREAMING_SNAKE_CASE__ = None	,SCREAMING_SNAKE_CASE__ = None	,SCREAMING_SNAKE_CASE__ = "pil"	,SCREAMING_SNAKE_CASE__ = True	,SCREAMING_SNAKE_CASE__ = None	,SCREAMING_SNAKE_CASE__ = 1	,)    ->   int:





             """simple docstring"""






             self.check_inputs(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							)

             # 2. Define call parameters
             if prompt is not None and isinstance(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							):
                   __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						1
             elif prompt is not None and isinstance(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							):
                   __SCREAMING_SNAKE_CASE     :int	       =						len(SCREAMING_SNAKE_CASE__							)
             else:
                   __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						prompt_embeds.shape[0]

             __SCREAMING_SNAKE_CASE     :Optional[int]	       =						self._execution_device
             # here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
             # of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
             # corresponds to doing no classifier free guidance.
             __SCREAMING_SNAKE_CASE     :int	       =						guidance_scale > 1.0

             # 3. Encode input prompt
             __SCREAMING_SNAKE_CASE     :Any	       =						self._encode_prompt(
                 SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,prompt_embeds=SCREAMING_SNAKE_CASE__	,negative_prompt_embeds=SCREAMING_SNAKE_CASE__	,)

             __SCREAMING_SNAKE_CASE     :Any	       =						ORT_TO_PT_TYPE[str(text_embeddings.dtype							)]

             # 4. Preprocess image
             __SCREAMING_SNAKE_CASE     :Dict	       =						preprocess(SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE     :str	       =						image.cpu()

             # 5. set timesteps
             self.scheduler.set_timesteps(SCREAMING_SNAKE_CASE__	,device=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE     :int	       =						self.scheduler.timesteps

             # 5. Add noise to image
             __SCREAMING_SNAKE_CASE     :Optional[int]	       =						torch.tensor([noise_level]	,dtype=torch.long	,device=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE     :str	       =						torch.randn(image.shape	,generator=SCREAMING_SNAKE_CASE__	,device=SCREAMING_SNAKE_CASE__	,dtype=SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE     :str	       =						self.low_res_scheduler.add_noise(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE     :Any	       =						2 if do_classifier_free_guidance else 1
             __SCREAMING_SNAKE_CASE     :Tuple	       =						np.concatenate([image] * batch_multiplier * num_images_per_prompt							)
             __SCREAMING_SNAKE_CASE     :Optional[int]	       =						np.concatenate([noise_level] * image.shape[0]							)

             # 6. Prepare latent variables
             __SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE     :Dict	       =						image.shape[2:]
             __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						self.prepare_latents(
                 batch_size * num_images_per_prompt	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,)

             # 7. Check that sizes of image and latents match
             __SCREAMING_SNAKE_CASE     :Dict	       =						image.shape[1]
             if NUM_LATENT_CHANNELS + num_channels_image != NUM_UNET_INPUT_CHANNELS:
                   raise ValueError(
                       '''Incorrect configuration settings! The config of `pipeline.unet` expects'''
                       f''' {NUM_UNET_INPUT_CHANNELS} but received `num_channels_latents`: {NUM_LATENT_CHANNELS} +'''
                       f''' `num_channels_image`: {num_channels_image} '''
                       f''' = {NUM_LATENT_CHANNELS+num_channels_image}. Please verify the config of'''
                       ''' `pipeline.unet` or your `image` input.'''							)

             # 8. Prepare extra step kwargs. TODO: Logic should ideally just be moved out of the pipeline
             __SCREAMING_SNAKE_CASE     :List[Any]	       =						self.prepare_extra_step_kwargs(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							)

             __SCREAMING_SNAKE_CASE     :Tuple	       =						next(
                 (input.type for input in self.unet.model.get_inputs() if input.name == '''timestep''')	,'''tensor(float)'''							)
             __SCREAMING_SNAKE_CASE     :List[str]	       =						ORT_TO_NP_TYPE[timestep_dtype]

             # 9. Denoising loop
             __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						len(SCREAMING_SNAKE_CASE__							) - num_inference_steps * self.scheduler.order
             with self.progress_bar(total=SCREAMING_SNAKE_CASE__							) as progress_bar:
                   for i, t in enumerate(SCREAMING_SNAKE_CASE__							):
                         # expand the latents if we are doing classifier free guidance
                         __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						np.concatenate([latents] * 2							) if do_classifier_free_guidance else latents

                         # concat latents, mask, masked_image_latents in the channel dimension
                         __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						self.scheduler.scale_model_input(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							)
                         __SCREAMING_SNAKE_CASE     :Tuple	       =						np.concatenate([latent_model_input, image]	,axis=1							)

                         # timestep to tensor
                         __SCREAMING_SNAKE_CASE     :Any	       =						np.array([t]	,dtype=SCREAMING_SNAKE_CASE__							)

                         # predict the noise residual
                         __SCREAMING_SNAKE_CASE     :Tuple	       =						self.unet(
                             sample=SCREAMING_SNAKE_CASE__	,timestep=SCREAMING_SNAKE_CASE__	,encoder_hidden_states=SCREAMING_SNAKE_CASE__	,class_labels=noise_level.astype(np.intaa							)	,)[0]

                         # perform guidance
                         if do_classifier_free_guidance:
                               __SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE     :int	       =						np.split(SCREAMING_SNAKE_CASE__	,2							)
                         __SCREAMING_SNAKE_CASE     :List[Any]	       =						noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

                         # compute the previous noisy sample x_t -> x_t-1
                         __SCREAMING_SNAKE_CASE     :Dict	       =						self.scheduler.step(
                             torch.from_numpy(SCREAMING_SNAKE_CASE__							)	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,**SCREAMING_SNAKE_CASE__							).prev_sample

                         # call the callback, if provided
                         if i == len(SCREAMING_SNAKE_CASE__							) - 1 or ((i + 1) > num_warmup_steps and (i + 1) % self.scheduler.order == 0):
                               progress_bar.update()
                               if callback is not None and i % callback_steps == 0:
                                     callback(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							)

        # 10. Post-processing
             __SCREAMING_SNAKE_CASE     :str	       =						self.decode_latents(latents.float()							)

             # 11. Convert to PIL
             if output_type == "pil":
                   __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						self.numpy_to_pil(SCREAMING_SNAKE_CASE__							)

             if not return_dict:
                   return (image,)

             return ImagePipelineOutput(images=SCREAMING_SNAKE_CASE__							)




       def 					_UpperCamelCase  (		self	,SCREAMING_SNAKE_CASE__							)    ->   str:





             """simple docstring"""






             __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						1 / 0.0_8_3_3_3 * latents
             __SCREAMING_SNAKE_CASE     :Tuple	       =						self.vae(latent_sample=SCREAMING_SNAKE_CASE__							)[0]
             __SCREAMING_SNAKE_CASE     :Dict	       =						np.clip(image / 2 + 0.5	,0	,1							)
             __SCREAMING_SNAKE_CASE     :List[str]	       =						image.transpose((0, 2, 3, 1)							)
             return image




       def 					_UpperCamelCase  (		self	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__ = None	,SCREAMING_SNAKE_CASE__ = None	,)    ->   Optional[Any]:





             """simple docstring"""






             if prompt is not None and isinstance(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							):
                   __SCREAMING_SNAKE_CASE     :List[str]	       =						1
             elif prompt is not None and isinstance(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							):
                   __SCREAMING_SNAKE_CASE     :Dict	       =						len(SCREAMING_SNAKE_CASE__							)
             else:
                   __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						prompt_embeds.shape[0]

             if prompt_embeds is None:
                   __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						self.tokenizer(
                       SCREAMING_SNAKE_CASE__	,padding='''max_length'''	,max_length=self.tokenizer.model_max_length	,truncation=SCREAMING_SNAKE_CASE__	,return_tensors='''pt'''	,)
                   __SCREAMING_SNAKE_CASE     :str	       =						text_inputs.input_ids
                   __SCREAMING_SNAKE_CASE     :List[Any]	       =						self.tokenizer(SCREAMING_SNAKE_CASE__	,padding='''longest'''	,return_tensors='''pt'''							).input_ids

                   if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(
                       SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							):
                         __SCREAMING_SNAKE_CASE     :Any	       =						self.tokenizer.batch_decode(
                             untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]							)
                         logger.warning(
                             '''The following part of your input was truncated because CLIP can only handle sequences up to'''
                             f''' {self.tokenizer.model_max_length} tokens: {removed_text}'''							)

                   # no positional arguments to text_encoder
                   __SCREAMING_SNAKE_CASE     :List[str]	       =						self.text_encoder(
                       input_ids=text_input_ids.int().to(SCREAMING_SNAKE_CASE__							)	,)
                   __SCREAMING_SNAKE_CASE     :List[str]	       =						prompt_embeds[0]

             __SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE       ,__SCREAMING_SNAKE_CASE     :Any	       =						prompt_embeds.shape
             # duplicate text embeddings for each generation per prompt, using mps friendly method
             __SCREAMING_SNAKE_CASE     :Tuple	       =						prompt_embeds.repeat(1	,SCREAMING_SNAKE_CASE__							)
             __SCREAMING_SNAKE_CASE     :Dict	       =						prompt_embeds.reshape(bs_embed * num_images_per_prompt	,SCREAMING_SNAKE_CASE__	,-1							)

             # get unconditional embeddings for classifier free guidance
             if do_classifier_free_guidance and negative_prompt_embeds is None:
                   __SCREAMING_SNAKE_CASE     :List[str]
                   if negative_prompt is None:
                         __SCREAMING_SNAKE_CASE     :Optional[int]	       =						[''''''] * batch_size
                   elif type(SCREAMING_SNAKE_CASE__							) is not type(SCREAMING_SNAKE_CASE__							):
                         raise TypeError(
                             f'''`negative_prompt` should be the same type to `prompt`, but got {type(SCREAMING_SNAKE_CASE__							)} !='''
                             f''' {type(SCREAMING_SNAKE_CASE__							)}.'''							)
                   elif isinstance(SCREAMING_SNAKE_CASE__	,SCREAMING_SNAKE_CASE__							):
                         __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						[negative_prompt]
                   elif batch_size != len(SCREAMING_SNAKE_CASE__							):
                         raise ValueError(
                             f'''`negative_prompt`: {negative_prompt} has batch size {len(SCREAMING_SNAKE_CASE__							)}, but `prompt`:'''
                             f''' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'''
                             ''' the batch size of `prompt`.'''							)
                   else:
                         __SCREAMING_SNAKE_CASE     :Any	       =						negative_prompt

                   __SCREAMING_SNAKE_CASE     :str	       =						text_input_ids.shape[-1]
                   __SCREAMING_SNAKE_CASE     :Dict	       =						self.tokenizer(
                       SCREAMING_SNAKE_CASE__	,padding='''max_length'''	,max_length=SCREAMING_SNAKE_CASE__	,truncation=SCREAMING_SNAKE_CASE__	,return_tensors='''pt'''	,)

                   # if hasattr(uncond_input, "attention_mask"):
                   #     attention_mask = uncond_input.attention_mask.to(device)
                   # else:
                   #     attention_mask = None

                   __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						self.text_encoder(
                       input_ids=uncond_input.input_ids.int().to(SCREAMING_SNAKE_CASE__							)	,)
                   __SCREAMING_SNAKE_CASE     :str	       =						uncond_embeddings[0]

             if do_classifier_free_guidance:
                   __SCREAMING_SNAKE_CASE     :Any	       =						uncond_embeddings.shape[1]
                   # duplicate unconditional embeddings for each generation per prompt, using mps friendly method
                   __SCREAMING_SNAKE_CASE     :Any	       =						uncond_embeddings.repeat(1	,SCREAMING_SNAKE_CASE__							)
                   __SCREAMING_SNAKE_CASE     :str	       =						uncond_embeddings.reshape(batch_size * num_images_per_prompt	,SCREAMING_SNAKE_CASE__	,-1							)

                   # For classifier free guidance, we need to do two forward passes.
                   # Here we concatenate the unconditional and text embeddings into a single batch
                   # to avoid doing two forward passes
                   __SCREAMING_SNAKE_CASE     :List[Any]	       =						np.concatenate([uncond_embeddings, prompt_embeds]							)

             return prompt_embeds






"""simple docstring"""
import argparse
import json

import gdown
import numpy as np
import torch
from huggingface_hub import hf_hub_download

from transformers import TimesformerConfig, TimesformerForVideoClassification, VideoMAEImageProcessor



def   __lowerCamelCase     (							a_  :	Union[str, Any]							)    ->   Tuple:
      __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						TimesformerConfig()

      if "large" in model_name:
            __SCREAMING_SNAKE_CASE     :Dict	       =						96

      if "hr" in model_name:
            __SCREAMING_SNAKE_CASE     :List[Any]	       =						16
            __SCREAMING_SNAKE_CASE     :int	       =						4_48

      __SCREAMING_SNAKE_CASE     :List[Any]	       =						'''huggingface/label-files'''
      if "k400" in model_name:
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						4_00
            __SCREAMING_SNAKE_CASE     :int	       =						'''kinetics400-id2label.json'''
      elif "k600" in model_name:
            __SCREAMING_SNAKE_CASE     :List[str]	       =						6_00
            __SCREAMING_SNAKE_CASE     :List[str]	       =						'''kinetics600-id2label.json'''
      elif "ssv2" in model_name:
            __SCREAMING_SNAKE_CASE     :Optional[int]	       =						1_74
            __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						'''something-something-v2-id2label.json'''
      else:
            raise ValueError('''Model name should either contain \'k400\', \'k600\' or \'ssv2\'.'''							)
      __SCREAMING_SNAKE_CASE     :int	       =						json.load(open(hf_hub_download(a_				,     a_				,     repo_type='''dataset'''							)				,     '''r'''							)							)
      __SCREAMING_SNAKE_CASE     :List[str]	       =						{int(a_							): v for k, v in idalabel.items()}
      __SCREAMING_SNAKE_CASE     :Optional[int]	       =						idalabel
      __SCREAMING_SNAKE_CASE     :Tuple	       =						{v: k for k, v in idalabel.items()}

      return config



def   __lowerCamelCase     (							a_  :	Optional[int]							)    ->   List[Any]:
      if "encoder." in name:
            __SCREAMING_SNAKE_CASE     :str	       =						name.replace('''encoder.'''				,     ''''''							)
      if "cls_token" in name:
            __SCREAMING_SNAKE_CASE     :int	       =						name.replace('''cls_token'''				,     '''timesformer.embeddings.cls_token'''							)
      if "pos_embed" in name:
            __SCREAMING_SNAKE_CASE     :str	       =						name.replace('''pos_embed'''				,     '''timesformer.embeddings.position_embeddings'''							)
      if "time_embed" in name:
            __SCREAMING_SNAKE_CASE     :List[str]	       =						name.replace('''time_embed'''				,     '''timesformer.embeddings.time_embeddings'''							)
      if "patch_embed.proj" in name:
            __SCREAMING_SNAKE_CASE     :Tuple	       =						name.replace('''patch_embed.proj'''				,     '''timesformer.embeddings.patch_embeddings.projection'''							)
      if "patch_embed.norm" in name:
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						name.replace('''patch_embed.norm'''				,     '''timesformer.embeddings.norm'''							)
      if "blocks" in name:
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						name.replace('''blocks'''				,     '''timesformer.encoder.layer'''							)
      if "attn.proj" in name:
            __SCREAMING_SNAKE_CASE     :Dict	       =						name.replace('''attn.proj'''				,     '''attention.output.dense'''							)
      if "attn" in name and "bias" not in name and "temporal" not in name:
            __SCREAMING_SNAKE_CASE     :List[Any]	       =						name.replace('''attn'''				,     '''attention.self'''							)
      if "attn" in name and "temporal" not in name:
            __SCREAMING_SNAKE_CASE     :Tuple	       =						name.replace('''attn'''				,     '''attention.attention'''							)
      if "temporal_norm1" in name:
            __SCREAMING_SNAKE_CASE     :Any	       =						name.replace('''temporal_norm1'''				,     '''temporal_layernorm'''							)
      if "temporal_attn.proj" in name:
            __SCREAMING_SNAKE_CASE     :int	       =						name.replace('''temporal_attn'''				,     '''temporal_attention.output.dense'''							)
      if "temporal_fc" in name:
            __SCREAMING_SNAKE_CASE     :Tuple	       =						name.replace('''temporal_fc'''				,     '''temporal_dense'''							)
      if "norm1" in name and "temporal" not in name:
            __SCREAMING_SNAKE_CASE     :List[Any]	       =						name.replace('''norm1'''				,     '''layernorm_before'''							)
      if "norm2" in name:
            __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						name.replace('''norm2'''				,     '''layernorm_after'''							)
      if "mlp.fc1" in name:
            __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						name.replace('''mlp.fc1'''				,     '''intermediate.dense'''							)
      if "mlp.fc2" in name:
            __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						name.replace('''mlp.fc2'''				,     '''output.dense'''							)
      if "norm.weight" in name and "fc" not in name and "temporal" not in name:
            __SCREAMING_SNAKE_CASE     :List[Any]	       =						name.replace('''norm.weight'''				,     '''timesformer.layernorm.weight'''							)
      if "norm.bias" in name and "fc" not in name and "temporal" not in name:
            __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						name.replace('''norm.bias'''				,     '''timesformer.layernorm.bias'''							)
      if "head" in name:
            __SCREAMING_SNAKE_CASE     :Dict	       =						name.replace('''head'''				,     '''classifier'''							)

      return name



def   __lowerCamelCase     (							a_  :	Optional[Any]				,     a_  :	Tuple							)    ->   Optional[int]:
      for key in orig_state_dict.copy().keys():
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						orig_state_dict.pop(a_							)

            if key.startswith('''model.'''							):
                  __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						key.replace('''model.'''				,     ''''''							)

            if "qkv" in key:
                  __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						key.split('''.'''							)
                  __SCREAMING_SNAKE_CASE     :List[str]	       =						int(key_split[1]							)
                  __SCREAMING_SNAKE_CASE     :str	       =						'''timesformer.encoder.layer.'''
                  if "temporal" in key:
                        __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						'''.temporal_attention.attention.qkv.'''
                  else:
                        __SCREAMING_SNAKE_CASE     :List[str]	       =						'''.attention.attention.qkv.'''
                  if "weight" in key:
                        __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						val
                  else:
                        __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						val
            else:
                  __SCREAMING_SNAKE_CASE     :Tuple	       =						val

      return orig_state_dict



def   __lowerCamelCase     (							)    ->   Optional[int]:
      __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						hf_hub_download(
          repo_id='''hf-internal-testing/spaghetti-video'''				,     filename='''eating_spaghetti.npy'''				,     repo_type='''dataset'''							)
      __SCREAMING_SNAKE_CASE     :Tuple	       =						np.load(a_							)
      return list(a_							)



def   __lowerCamelCase     (							a_  :	List[Any]				,     a_  :	int				,     a_  :	Optional[int]				,     a_  :	Optional[Any]							)    ->   Union[str, Any]:
      __SCREAMING_SNAKE_CASE     :List[Any]	       =						get_timesformer_config(a_							)

      __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						TimesformerForVideoClassification(a_							)

      # download original checkpoint, hosted on Google Drive
      __SCREAMING_SNAKE_CASE     :List[Any]	       =						'''pytorch_model.bin'''
      gdown.cached_download(a_				,     a_				,     quiet=a_							)
      __SCREAMING_SNAKE_CASE     :Tuple	       =						torch.load(a_				,     map_location='''cpu'''							)
      if "model" in files:
            __SCREAMING_SNAKE_CASE     :List[str]	       =						files['''model''']
      elif "module" in files:
            __SCREAMING_SNAKE_CASE     :Tuple	       =						files['''module''']
      else:
            __SCREAMING_SNAKE_CASE     :Optional[int]	       =						files['''model_state''']
      __SCREAMING_SNAKE_CASE     :Any	       =						convert_state_dict(a_				,     a_							)

      model.load_state_dict(a_							)
      model.eval()

      # verify model on basic input
      __SCREAMING_SNAKE_CASE     :Tuple	       =						VideoMAEImageProcessor(image_mean=[0.5, 0.5, 0.5]				,     image_std=[0.5, 0.5, 0.5]							)
      __SCREAMING_SNAKE_CASE     :str	       =						prepare_video()
      __SCREAMING_SNAKE_CASE     :Tuple	       =						image_processor(video[:8]				,     return_tensors='''pt'''							)

      __SCREAMING_SNAKE_CASE     :Tuple	       =						model(**a_							)
      __SCREAMING_SNAKE_CASE     :List[str]	       =						outputs.logits

      __SCREAMING_SNAKE_CASE     :Dict	       =						[
          # Kinetics-400 checkpoints (hr = high resolution input of 448px instead of 224px)
          '''timesformer-base-finetuned-k400''',
          '''timesformer-large-finetuned-k400''',
          '''timesformer-hr-finetuned-k400''',
          # Kinetics-600 checkpoints (hr = high resolution input of 448px instead of 224px)
          '''timesformer-base-finetuned-k600''',
          '''timesformer-large-finetuned-k600''',
          '''timesformer-hr-finetuned-k600''',
          # Something-Something-v2 checkpoints (hr = high resolution input of 448px instead of 224px)
          '''timesformer-base-finetuned-ssv2''',
          '''timesformer-large-finetuned-ssv2''',
          '''timesformer-hr-finetuned-ssv2''',
      ]

      # NOTE: logits were tested with image_mean and image_std equal to [0.5, 0.5, 0.5] and [0.5, 0.5, 0.5]
      if model_name == "timesformer-base-finetuned-k400":
            __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						torch.Size([1, 4_00]							)
            __SCREAMING_SNAKE_CASE     :int	       =						torch.tensor([-0.3016, -0.7713, -0.4205]							)
      elif model_name == "timesformer-base-finetuned-k600":
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						torch.Size([1, 6_00]							)
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						torch.tensor([-0.7267, -0.7466, 3.2404]							)
      elif model_name == "timesformer-base-finetuned-ssv2":
            __SCREAMING_SNAKE_CASE     :List[str]	       =						torch.Size([1, 1_74]							)
            __SCREAMING_SNAKE_CASE     :Tuple	       =						torch.tensor([-0.9059, 0.6433, -3.1457]							)
      elif model_name == "timesformer-large-finetuned-k400":
            __SCREAMING_SNAKE_CASE     :Tuple	       =						torch.Size([1, 4_00]							)
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						torch.tensor([0, 0, 0]							)
      elif model_name == "timesformer-large-finetuned-k600":
            __SCREAMING_SNAKE_CASE     :List[Any]	       =						torch.Size([1, 6_00]							)
            __SCREAMING_SNAKE_CASE     :str	       =						torch.tensor([0, 0, 0]							)
      elif model_name == "timesformer-large-finetuned-ssv2":
            __SCREAMING_SNAKE_CASE     :int	       =						torch.Size([1, 1_74]							)
            __SCREAMING_SNAKE_CASE     :Any	       =						torch.tensor([0, 0, 0]							)
      elif model_name == "timesformer-hr-finetuned-k400":
            __SCREAMING_SNAKE_CASE     :Union[str, Any]	       =						torch.Size([1, 4_00]							)
            __SCREAMING_SNAKE_CASE     :Dict	       =						torch.tensor([-0.9617, -3.7311, -3.7708]							)
      elif model_name == "timesformer-hr-finetuned-k600":
            __SCREAMING_SNAKE_CASE     :int	       =						torch.Size([1, 6_00]							)
            __SCREAMING_SNAKE_CASE     :Optional[Any]	       =						torch.tensor([2.5273, 0.7127, 1.8848]							)
      elif model_name == "timesformer-hr-finetuned-ssv2":
            __SCREAMING_SNAKE_CASE     :List[Any]	       =						torch.Size([1, 1_74]							)
            __SCREAMING_SNAKE_CASE     :List[Any]	       =						torch.tensor([-3.6756, -0.7513, 0.7180]							)
      else:
            raise ValueError(f'''Model name not supported. Should be one of {model_names}'''							)

      # verify logits
      assert logits.shape == expected_shape
      assert torch.allclose(logits[0, :3]				,     a_				,     atol=1e-4							)
      print('''Logits ok!'''							)

      if pytorch_dump_folder_path is not None:
            print(f'''Saving model and image processor to {pytorch_dump_folder_path}'''							)
            image_processor.save_pretrained(a_							)
            model.save_pretrained(a_							)

      if push_to_hub:
            print('''Pushing to the hub...'''							)
            model.push_to_hub(f'''fcakyon/{model_name}'''							)


if __name__ == "__main__":
  lowerCamelCase_          =	argparse.ArgumentParser()
  # Required parameters
  parser.add_argument(
      "--checkpoint_url",
      default="https://drive.google.com/u/1/uc?id=17yvuYp9L4mn-HpIcK5Zo6K3UoOy1kA5l&export=download",
      type=str,
      help=(
          "URL of the original PyTorch checkpoint (on Google Drive) you'd like to convert. Should be a direct"
          " download link."
      ),
  )
  parser.add_argument(
      "--pytorch_dump_folder_path",
      default="",
      type=str,
      help="Path to the output PyTorch model directory.",
  )
  parser.add_argument("--model_name", default="timesformer-base-finetuned-k400", type=str, help="Name of the model.")
  parser.add_argument(
      "--push_to_hub", action="store_true", help="Whether or not to push the converted model to the 🤗 hub."
  )

  lowerCamelCase_          =	parser.parse_args()
  convert_timesformer_checkpoint(
      args.checkpoint_url, args.pytorch_dump_folder_path, args.model_name, args.push_to_hub
  )