

import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    UNetaDConditionModel,
    VideoToVideoSDPipeline,
)
from diffusers.utils import floats_tensor, is_xformers_available, skip_mps
from diffusers.utils.testing_utils import enable_full_determinism, slow, torch_device

from ..pipeline_params import (
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()





@skip_mps
class A__    ( A  ,		unittest.TestCase ):



 """simple docstring"""







 _lowercase     :					int      =						VideoToVideoSDPipeline
 _lowercase     :					Tuple      =						TEXT_GUIDED_IMAGE_VARIATION_PARAMS.union({'''video'''} ) - {'''image''', '''width''', '''height'''}
 _lowercase     :					List[str]      =						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({'''video'''} ) - {'''image'''}
 _lowercase     :					Union[str, Any]      =						PipelineTesterMixin.required_optional_params - {'''latents'''}
 _lowercase     :					List[Any]      =						False

 # No `output_type`.
 _lowercase     :					Optional[int]      =						frozenset(
     [
         '''num_inference_steps''',
         '''generator''',
         '''latents''',
         '''return_dict''',
         '''callback''',
         '''callback_steps''',
     ] )

 def    __magic_name__       (		self :	Optional[int]       ):






      '''simple docstring'''




      torch.manual_seed(0       )
      _lowerCAmelCase   :					Union[str, Any]        =    UNetaDConditionModel(
          block_out_channels=(3_2, 6_4, 6_4, 6_4)     ,						layers_per_block=2     ,						sample_size=3_2     ,						in_channels=4     ,						out_channels=4     ,						down_block_types=("CrossAttnDownBlock3D", "CrossAttnDownBlock3D", "CrossAttnDownBlock3D", "DownBlock3D")     ,						up_block_types=("UpBlock3D", "CrossAttnUpBlock3D", "CrossAttnUpBlock3D", "CrossAttnUpBlock3D")     ,						cross_attention_dim=3_2     ,						attention_head_dim=4     ,						)
      _lowerCAmelCase   :					Tuple        =    DDIMScheduler(
          beta_start=0.00085     ,						beta_end=0.012     ,						beta_schedule="scaled_linear"     ,						clip_sample=A_     ,						set_alpha_to_one=A_     ,						)
      torch.manual_seed(0       )
      _lowerCAmelCase   :					Union[str, Any]        =    AutoencoderKL(
          block_out_channels=[3_2, 6_4]     ,						in_channels=3     ,						out_channels=3     ,						down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]     ,						up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]     ,						latent_channels=4     ,						sample_size=1_2_8     ,						)
      torch.manual_seed(0       )
      _lowerCAmelCase   :					str        =    CLIPTextConfig(
          bos_token_id=0     ,						eos_token_id=2     ,						hidden_size=3_2     ,						intermediate_size=3_7     ,						layer_norm_eps=1E-05     ,						num_attention_heads=4     ,						num_hidden_layers=5     ,						pad_token_id=1     ,						vocab_size=1_0_0_0     ,						hidden_act="gelu"     ,						projection_dim=5_1_2     ,						)
      _lowerCAmelCase   :					Union[str, Any]        =    CLIPTextModel(A_       )
      _lowerCAmelCase   :					Union[str, Any]        =    CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"       )

      _lowerCAmelCase   :					int        =    {
          "unet": unet,
          "scheduler": scheduler,
          "vae": vae,
          "text_encoder": text_encoder,
          "tokenizer": tokenizer,
      }
      return components

 def    __magic_name__       (		self :	str     ,						A_ :	Union[str, Any]     ,						A_ :	List[Any]=0       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Dict        =    floats_tensor((1, 3, 3, 3_2, 3_2)     ,						rng=random.Random(A_       )       ).to(A_       )

      if str(A_       ).startswith("mps"       ):
           _lowerCAmelCase   :					str        =    torch.manual_seed(A_       )
      else:
           _lowerCAmelCase   :					List[str]        =    torch.Generator(device=A_       ).manual_seed(A_       )
      _lowerCAmelCase   :					Optional[Any]        =    {
          "prompt": "A painting of a squirrel eating a burger",
          "video": video,
          "generator": generator,
          "num_inference_steps": 2,
          "guidance_scale": 6.0,
          "output_type": "pt",
      }
      return inputs

 def    __magic_name__       (		self :	List[Any]       ):






      '''simple docstring'''




      _lowerCAmelCase   :					int        =    "cpu"  # ensure determinism for the device-dependent torch.Generator
      _lowerCAmelCase   :					str        =    self.get_dummy_components()
      _lowerCAmelCase   :					Optional[Any]        =    VideoToVideoSDPipeline(**A_       )
      _lowerCAmelCase   :					Dict        =    sd_pipe.to(A_       )
      sd_pipe.set_progress_bar_config(disable=A_       )

      _lowerCAmelCase   :					Union[str, Any]        =    self.get_dummy_inputs(A_       )
      _lowerCAmelCase   :					Union[str, Any]        =    "np"
      _lowerCAmelCase   :					Optional[int]        =    sd_pipe(**A_       ).frames
      _lowerCAmelCase   :					Optional[int]        =    frames[0][-3:, -3:, -1]

      assert frames[0].shape == (3_2, 3_2, 3)
      _lowerCAmelCase   :					Optional[int]        =    np.array([1_0_6, 1_1_7, 1_1_3, 1_7_4, 1_3_7, 1_1_2, 1_4_8, 1_5_1, 1_3_1]       )

      assert np.abs(image_slice.flatten() - expected_slice       ).max() < 1E-2

 @unittest.skipIf(
     torch_device != "cuda" or not is_xformers_available()     ,						reason="XFormers attention is only available with CUDA and `xformers` installed"     ,						)
 def    __magic_name__       (		self :	Dict       ):






      '''simple docstring'''




      self._test_xformers_attention_forwardGenerator_pass(test_mean_pixel_difference=A_     ,						expected_max_diff=5E-3       )

 @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."       )
 def    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




      pass

 @unittest.skip(reason="Batching needs to be properly figured out first for this pipeline."       )
 def    __magic_name__       (		self :	Tuple       ):






      '''simple docstring'''




      pass

 @unittest.skip(reason="`num_images_per_prompt` argument is not supported for this pipeline."       )
 def    __magic_name__       (		self :	Union[str, Any]       ):






      '''simple docstring'''




      pass


 def    __magic_name__       (		self :	List[str]       ):






      '''simple docstring'''




      return super().test_progress_bar()





@slow
@skip_mps
class A__    ( unittest.TestCase ):



 """simple docstring"""







 def    __magic_name__       (		self :	Any       ):






      '''simple docstring'''




      _lowerCAmelCase   :					Tuple        =    VideoToVideoSDPipeline.from_pretrained("cerspense/zeroscope_v2_XL"     ,						torch_dtype=torch.floataa       )
      pipe.enable_model_cpu_offload()

      # 10 frames
      _lowerCAmelCase   :					List[Any]        =    torch.Generator(device="cpu"       ).manual_seed(0       )
      _lowerCAmelCase   :					str        =    torch.randn((1, 1_0, 3, 1_0_2_4, 5_7_6)     ,						generator=A_       )
      _lowerCAmelCase   :					int        =    video.to("cuda"       )

      _lowerCAmelCase   :					Tuple        =    "Spiderman is surfing"

      _lowerCAmelCase   :					str        =    pipe(A_     ,						video=A_     ,						generator=A_     ,						num_inference_steps=3     ,						output_type="pt"       ).frames

      _lowerCAmelCase   :					int        =    np.array([-1.0458984, -1.1279297, -0.9663086, -0.91503906, -0.75097656]       )
      assert np.abs(video_frames.cpu().numpy()[0, 0, 0, 0, -5:] - expected_array       ).sum() < 1E-2









import argparse
import os

import torch

from transformers import FlavaConfig, FlavaForPreTraining
from transformers.models.flava.convert_dalle_to_flava_codebook import convert_dalle_checkpoint







def 	_snake_case		(    SCREAMING_SNAKE_CASE				)       ->      Optional[int]:

     """simple docstring"""







     return sum(param.float().sum() if "encoder.embeddings" not in key else 0 for key, param in state_dict.items()				)







def 	_snake_case		(    SCREAMING_SNAKE_CASE		,							SCREAMING_SNAKE_CASE				)       ->      Tuple:

     """simple docstring"""







     _lowerCAmelCase   :					str        =    {}

     for key, value in state_dict.items():
          if "text_encoder.embeddings" in key or "image_encoder.embeddings" in key:
               continue

          _lowerCAmelCase   :					Union[str, Any]        =    key.replace("heads.cmd.mim_head.cls.predictions"		,							"mmm_image_head"				)
          _lowerCAmelCase   :					int        =    key.replace("heads.cmd.mlm_head.cls.predictions"		,							"mmm_text_head"				)
          _lowerCAmelCase   :					Any        =    key.replace("heads.cmd.itm_head.cls"		,							"itm_head"				)
          _lowerCAmelCase   :					Optional[Any]        =    key.replace("heads.cmd.itm_head.pooler"		,							"itm_head.pooler"				)
          _lowerCAmelCase   :					Any        =    key.replace("heads.cmd.clip_head.logit_scale"		,							"flava.logit_scale"				)
          _lowerCAmelCase   :					List[Any]        =    key.replace("heads.fairseq_mlm.cls.predictions"		,							"mlm_head"				)
          _lowerCAmelCase   :					Union[str, Any]        =    key.replace("heads.imagenet.mim_head.cls.predictions"		,							"mim_head"				)
          _lowerCAmelCase   :					Optional[int]        =    key.replace("mm_text_projection"		,							"flava.text_to_mm_projection"				)
          _lowerCAmelCase   :					int        =    key.replace("mm_image_projection"		,							"flava.image_to_mm_projection"				)
          _lowerCAmelCase   :					List[str]        =    key.replace("image_encoder.module"		,							"flava.image_model"				)
          _lowerCAmelCase   :					Optional[int]        =    key.replace("text_encoder.module"		,							"flava.text_model"				)
          _lowerCAmelCase   :					List[str]        =    key.replace("mm_encoder.module.encoder.cls_token"		,							"flava.multimodal_model.cls_token"				)
          _lowerCAmelCase   :					Dict        =    key.replace("mm_encoder.module"		,							"flava.multimodal_model"				)
          _lowerCAmelCase   :					Dict        =    key.replace("text_projection"		,							"flava.text_projection"				)
          _lowerCAmelCase   :					Optional[Any]        =    key.replace("image_projection"		,							"flava.image_projection"				)

          _lowerCAmelCase   :					List[Any]        =    value.float()

     for key, value in codebook_state_dict.items():
          _lowerCAmelCase   :					int        =    value

     return upgrade







@torch.no_grad()
def 	_snake_case		(    SCREAMING_SNAKE_CASE		,							SCREAMING_SNAKE_CASE		,							SCREAMING_SNAKE_CASE		,							SCREAMING_SNAKE_CASE=None				)       ->      Tuple:

     """simple docstring"""







     if config_path is not None:
          _lowerCAmelCase   :					Dict        =    FlavaConfig.from_pretrained(SCREAMING_SNAKE_CASE				)
     else:
          _lowerCAmelCase   :					Dict        =    FlavaConfig()

     _lowerCAmelCase   :					List[Any]        =    FlavaForPreTraining(SCREAMING_SNAKE_CASE				).eval()

     _lowerCAmelCase   :					Optional[int]        =    convert_dalle_checkpoint(SCREAMING_SNAKE_CASE		,							SCREAMING_SNAKE_CASE		,							save_checkpoint=SCREAMING_SNAKE_CASE				)

     if os.path.exists(SCREAMING_SNAKE_CASE				):
          _lowerCAmelCase   :					Dict        =    torch.load(SCREAMING_SNAKE_CASE		,							map_location="cpu"				)
     else:
          _lowerCAmelCase   :					Tuple        =    torch.hub.load_state_dict_from_url(SCREAMING_SNAKE_CASE		,							map_location="cpu"				)

     _lowerCAmelCase   :					Optional[int]        =    upgrade_state_dict(SCREAMING_SNAKE_CASE		,							SCREAMING_SNAKE_CASE				)
     hf_model.load_state_dict(SCREAMING_SNAKE_CASE				)
     _lowerCAmelCase   :					Optional[Any]        =    hf_model.state_dict()
     _lowerCAmelCase   :					Optional[Any]        =    count_parameters(SCREAMING_SNAKE_CASE				)
     _lowerCAmelCase   :					Optional[Any]        =    count_parameters(SCREAMING_SNAKE_CASE				) + count_parameters(SCREAMING_SNAKE_CASE				)

     assert torch.allclose(SCREAMING_SNAKE_CASE		,							SCREAMING_SNAKE_CASE		,							atol=1e-3				)

     hf_model.save_pretrained(SCREAMING_SNAKE_CASE				)


if __name__ == "__main__":
  __UpperCAmelCase    				=     argparse.ArgumentParser()
  parser.add_argument('--pytorch_dump_folder_path', default=None, type=str, help='Path to the output PyTorch model.')
  parser.add_argument('--checkpoint_path', default=None, type=str, help='Path to flava checkpoint')
  parser.add_argument('--codebook_path', default=None, type=str, help='Path to flava codebook checkpoint')
  parser.add_argument('--config_path', default=None, type=str, help='Path to hf config.json of model to convert')
  __UpperCAmelCase    				=     parser.parse_args()

  convert_flava_checkpoint(args.checkpoint_path, args.codebook_path, args.pytorch_dump_folder_path, args.config_path)






