






import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image

from diffusers import (
    DDIMScheduler,
    KandinskyVaaImgaImgPipeline,
    KandinskyVaaPriorPipeline,
    UNetaDConditionModel,
    VQModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()







class 		_A						( UpperCAmelCase_	,				unittest.TestCase ):
  lowercase_						:			str						=					KandinskyVaaImgaImgPipeline
  lowercase_						:			List[Any]						=					['''image_embeds''', '''negative_image_embeds''', '''image''']
  lowercase_						:			Optional[Any]						=					[
      '''image_embeds''',
      '''negative_image_embeds''',
      '''image''',
  ]
  lowercase_						:			List[str]						=					[
      '''generator''',
      '''height''',
      '''width''',
      '''strength''',
      '''guidance_scale''',
      '''num_inference_steps''',
      '''return_dict''',
      '''guidance_scale''',
      '''num_images_per_prompt''',
      '''output_type''',
      '''return_dict''',
  ]
  lowercase_						:			int						=					False

  @property
  def    a					(    self						:	Union[str, Any]			):
     """simple docstring"""

     return 32

  @property
  def    a					(    self						:	Union[str, Any]			):
     """simple docstring"""

     return 32

  @property
  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     return self.time_input_dim

  @property
  def    a					(    self						:	int			):
     """simple docstring"""

     return self.time_input_dim * 4

  @property
  def    a					(    self						:	Union[str, Any]			):
     """simple docstring"""

     return 1_00

  @property
  def    a					(    self						:	Tuple			):
     """simple docstring"""

     torch.manual_seed(0			)

     __UpperCamelCase	:			Tuple					=					{
         """in_channels""": 4,
         # Out channels is double in channels because predicts mean and variance
         """out_channels""": 8,
         """addition_embed_type""": """image""",
         """down_block_types""": ("""ResnetDownsampleBlock2D""", """SimpleCrossAttnDownBlock2D"""),
         """up_block_types""": ("""SimpleCrossAttnUpBlock2D""", """ResnetUpsampleBlock2D"""),
         """mid_block_type""": """UNetMidBlock2DSimpleCrossAttn""",
         """block_out_channels""": (self.block_out_channels_a, self.block_out_channels_a * 2),
         """layers_per_block""": 1,
         """encoder_hid_dim""": self.text_embedder_hidden_size,
         """encoder_hid_dim_type""": """image_proj""",
         """cross_attention_dim""": self.cross_attention_dim,
         """attention_head_dim""": 4,
         """resnet_time_scale_shift""": """scale_shift""",
         """class_embed_type""": None,
     }

     __UpperCamelCase	:			Dict					=					UNetaDConditionModel(**lowerCamelCase__			)
     return model

  @property
  def    a					(    self						:	Tuple			):
     """simple docstring"""

     return {
         "block_out_channels": [32, 64],
         "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
         "in_channels": 3,
         "latent_channels": 4,
         "layers_per_block": 1,
         "norm_num_groups": 8,
         "norm_type": "spatial",
         "num_vq_embeddings": 12,
         "out_channels": 3,
         "up_block_types": [
             "AttnUpDecoderBlock2D",
             "UpDecoderBlock2D",
         ],
         "vq_embed_dim": 4,
     }

  @property
  def    a					(    self						:	Tuple			):
     """simple docstring"""

     torch.manual_seed(0			)
     __UpperCamelCase	:			Optional[Any]					=					VQModel(**self.dummy_movq_kwargs			)
     return model

  def    a					(    self						:	Optional[Any]			):
     """simple docstring"""

     __UpperCamelCase	:			int					=					self.dummy_unet
     __UpperCamelCase	:			List[Any]					=					self.dummy_movq

     __UpperCamelCase	:			Union[str, Any]					=					{
         """num_train_timesteps""": 10_00,
         """beta_schedule""": """linear""",
         """beta_start""": 0.0_0085,
         """beta_end""": 0.012,
         """clip_sample""": False,
         """set_alpha_to_one""": False,
         """steps_offset""": 0,
         """prediction_type""": """epsilon""",
         """thresholding""": False,
     }

     __UpperCamelCase	:			Union[str, Any]					=					DDIMScheduler(**lowerCamelCase__			)

     __UpperCamelCase	:			List[Any]					=					{
         """unet""": unet,
         """scheduler""": scheduler,
         """movq""": movq,
     }

     return components

  def    a					(    self						:	List[Any]   ,				lowerCamelCase__						:	Dict   ,				lowerCamelCase__						:	Optional[int]=0			):
     """simple docstring"""

     __UpperCamelCase	:			Any					=					floats_tensor((1, self.text_embedder_hidden_size)   ,				rng=random.Random(lowerCamelCase__			)			).to(lowerCamelCase__			)
     __UpperCamelCase	:			Dict					=					floats_tensor((1, self.text_embedder_hidden_size)   ,				rng=random.Random(seed + 1			)			).to(
         lowerCamelCase__			)
     # create init_image
     __UpperCamelCase	:			Optional[int]					=					floats_tensor((1, 3, 64, 64)   ,				rng=random.Random(lowerCamelCase__			)			).to(lowerCamelCase__			)
     __UpperCamelCase	:			List[str]					=					image.cpu().permute(0   ,				2   ,				3   ,				1			)[0]
     __UpperCamelCase	:			Dict					=					Image.fromarray(np.uinta(lowerCamelCase__			)			).convert("""RGB"""			).resize((2_56, 2_56)			)

     if str(lowerCamelCase__			).startswith("""mps"""			):
        __UpperCamelCase	:			Tuple					=					torch.manual_seed(lowerCamelCase__			)
     else:
        __UpperCamelCase	:			List[str]					=					torch.Generator(device=lowerCamelCase__			).manual_seed(lowerCamelCase__			)
     __UpperCamelCase	:			Dict					=					{
         """image""": init_image,
         """image_embeds""": image_embeds,
         """negative_image_embeds""": negative_image_embeds,
         """generator""": generator,
         """height""": 64,
         """width""": 64,
         """num_inference_steps""": 10,
         """guidance_scale""": 7.0,
         """strength""": 0.2,
         """output_type""": """np""",
     }
     return inputs

  def    a					(    self						:	str			):
     """simple docstring"""

     __UpperCamelCase	:			Optional[int]					=					"""cpu"""

     __UpperCamelCase	:			List[Any]					=					self.get_dummy_components()

     __UpperCamelCase	:			List[Any]					=					self.pipeline_class(**lowerCamelCase__			)
     __UpperCamelCase	:			Any					=					pipe.to(lowerCamelCase__			)

     pipe.set_progress_bar_config(disable=lowerCamelCase__			)

     __UpperCamelCase	:			Union[str, Any]					=					pipe(**self.get_dummy_inputs(lowerCamelCase__			)			)
     __UpperCamelCase	:			Tuple					=					output.images

     __UpperCamelCase	:			str					=					pipe(
         **self.get_dummy_inputs(lowerCamelCase__			)   ,				return_dict=lowerCamelCase__   ,				)[0]

     __UpperCamelCase	:			Optional[Any]					=					image[0, -3:, -3:, -1]
     __UpperCamelCase	:			int					=					image_from_tuple[0, -3:, -3:, -1]

     assert image.shape == (1, 64, 64, 3)

     __UpperCamelCase	:			List[str]					=					np.array(
         [0.619_9778, 0.6398_4406, 0.4614_5785, 0.6294_4984, 0.562_2215, 0.4730_6132, 0.4744_1456, 0.460_7606, 0.4871_9263]			)
     assert (
         np.abs(image_slice.flatten() - expected_slice			).max() < 1e-2
     ), f' expected_slice {expected_slice}, but got {image_slice.flatten()}'
     assert (
         np.abs(image_from_tuple_slice.flatten() - expected_slice			).max() < 1e-2
     ), f' expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}'







@slow
@require_torch_gpu
class 		_A						( unittest.TestCase ):

  def    a					(    self						:	List[str]			):
     """simple docstring"""

     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()

  def    a					(    self						:	Optional[int]			):
     """simple docstring"""

     __UpperCamelCase	:			int					=					load_numpy(
         """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
         """/kandinskyv22/kandinskyv22_img2img_frog.npy"""			)

     __UpperCamelCase	:			Optional[Any]					=					load_image(
         """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main""" """/kandinsky/cat.png"""			)
     __UpperCamelCase	:			str					=					"""A red cartoon frog, 4k"""

     __UpperCamelCase	:			List[Any]					=					KandinskyVaaPriorPipeline.from_pretrained(
         """kandinsky-community/kandinsky-2-2-prior"""   ,				torch_dtype=torch.floataa			)
     pipe_prior.to(lowerCamelCase__			)

     __UpperCamelCase	:			int					=					KandinskyVaaImgaImgPipeline.from_pretrained(
         """kandinsky-community/kandinsky-2-2-decoder"""   ,				torch_dtype=torch.floataa			)
     __UpperCamelCase	:			Optional[Any]					=					pipeline.to(lowerCamelCase__			)

     pipeline.set_progress_bar_config(disable=lowerCamelCase__			)

     __UpperCamelCase	:			List[str]					=					torch.Generator(device="""cpu"""			).manual_seed(0			)
     __UpperCamelCase	,     __UpperCamelCase	:			Any					=					pipe_prior(
         lowerCamelCase__   ,				generator=lowerCamelCase__   ,				num_inference_steps=5   ,				negative_prompt=""""""   ,				).to_tuple()

     __UpperCamelCase	:			int					=					pipeline(
         image=lowerCamelCase__   ,				image_embeds=lowerCamelCase__   ,				negative_image_embeds=lowerCamelCase__   ,				generator=lowerCamelCase__   ,				num_inference_steps=1_00   ,				height=7_68   ,				width=7_68   ,				strength=0.2   ,				output_type="""np"""   ,				)

     __UpperCamelCase	:			Dict					=					output.images[0]

     assert image.shape == (7_68, 7_68, 3)

     assert_mean_pixel_difference(lowerCamelCase__   ,				lowerCamelCase__			)











import argparse
from pathlib import Path
from typing import Dict, OrderedDict, Tuple

import torch
from audiocraft.models import MusicGen

from transformers import (
    AutoFeatureExtractor,
    AutoTokenizer,
    EncodecModel,
    MusicgenDecoderConfig,
    MusicgenForConditionalGeneration,
    MusicgenProcessor,
    TaEncoderModel,
)
from transformers.models.musicgen.modeling_musicgen import MusicgenForCausalLM
from transformers.utils import logging


logging.set_verbosity_info()
UpperCamelCase    							=      logging.get_logger(__name__)


UpperCamelCase    							=      ['model.decoder.embed_positions.weights']


def 				__lowerCamelCase     (							__lowerCAmelCase							:					Optional[int]							)     ->  Union[str, Any]:
   if "emb" in name:
      __UpperCamelCase	:			Any					=					name.replace("""emb"""					,							"""model.decoder.embed_tokens"""							)
   if "transformer" in name:
      __UpperCamelCase	:			str					=					name.replace("""transformer"""					,							"""model.decoder"""							)
   if "cross_attention" in name:
      __UpperCamelCase	:			List[str]					=					name.replace("""cross_attention"""					,							"""encoder_attn"""							)
   if "linear1" in name:
      __UpperCamelCase	:			Optional[int]					=					name.replace("""linear1"""					,							"""fc1"""							)
   if "linear2" in name:
      __UpperCamelCase	:			Optional[Any]					=					name.replace("""linear2"""					,							"""fc2"""							)
   if "norm1" in name:
      __UpperCamelCase	:			Tuple					=					name.replace("""norm1"""					,							"""self_attn_layer_norm"""							)
   if "norm_cross" in name:
      __UpperCamelCase	:			List[str]					=					name.replace("""norm_cross"""					,							"""encoder_attn_layer_norm"""							)
   if "norm2" in name:
      __UpperCamelCase	:			Tuple					=					name.replace("""norm2"""					,							"""final_layer_norm"""							)
   if "out_norm" in name:
      __UpperCamelCase	:			Union[str, Any]					=					name.replace("""out_norm"""					,							"""model.decoder.layer_norm"""							)
   if "linears" in name:
      __UpperCamelCase	:			Optional[int]					=					name.replace("""linears"""					,							"""lm_heads"""							)
   if "condition_provider.conditioners.description.output_proj" in name:
      __UpperCamelCase	:			Tuple					=					name.replace("""condition_provider.conditioners.description.output_proj"""					,							"""enc_to_dec_proj"""							)
   return name


def 				__lowerCamelCase     (							__lowerCAmelCase							:					OrderedDict					,							__lowerCAmelCase							:					int							)     ->  Tuple[Dict, Dict]:
   __UpperCamelCase	:			Tuple					=					list(state_dict.keys()							)
   __UpperCamelCase	:			List[Any]					=					{}
   for key in keys:
      __UpperCamelCase	:			Optional[Any]					=					state_dict.pop(__lowerCAmelCase							)
      __UpperCamelCase	:			Dict					=					rename_keys(__lowerCAmelCase							)
      if "in_proj_weight" in key:
         # split fused qkv proj
         __UpperCamelCase	:			Optional[Any]					=					val[:hidden_size, :]
         __UpperCamelCase	:			Optional[Any]					=					val[hidden_size : 2 * hidden_size, :]
         __UpperCamelCase	:			List[str]					=					val[-hidden_size:, :]
      elif "enc_to_dec_proj" in key:
         __UpperCamelCase	:			Dict					=					val
      else:
         __UpperCamelCase	:			Any					=					val
   return state_dict, enc_dec_proj_state_dict


def 				__lowerCamelCase     (							__lowerCAmelCase							:					str							)     ->  MusicgenDecoderConfig:
   if checkpoint == "small":
      # default config values
      __UpperCamelCase	:			int					=					1024
      __UpperCamelCase	:			Union[str, Any]					=					24
      __UpperCamelCase	:			int					=					16
   elif checkpoint == "medium":
      __UpperCamelCase	:			List[Any]					=					1536
      __UpperCamelCase	:			Dict					=					48
      __UpperCamelCase	:			Dict					=					24
   elif checkpoint == "large":
      __UpperCamelCase	:			List[Any]					=					2048
      __UpperCamelCase	:			str					=					48
      __UpperCamelCase	:			Optional[int]					=					32
   else:
      raise ValueError(f'Checkpoint should be one of `[\'small\', \'medium\', \'large\']`, got {checkpoint}.'							)
   __UpperCamelCase	:			Any					=					MusicgenDecoderConfig(
       hidden_size=__lowerCAmelCase					,							ffn_dim=hidden_size * 4					,							num_hidden_layers=__lowerCAmelCase					,							num_attention_heads=__lowerCAmelCase					,							)
   return config


@torch.no_grad()
def 				__lowerCamelCase     (							__lowerCAmelCase							:					List[str]					,							__lowerCAmelCase							:					Optional[Any]=None					,							__lowerCAmelCase							:					Tuple=None					,							__lowerCAmelCase							:					str="cpu"							)     ->  Optional[int]:
   __UpperCamelCase	:			str					=					MusicGen.get_pretrained(__lowerCAmelCase					,							device=__lowerCAmelCase							)
   __UpperCamelCase	:			int					=					decoder_config_from_checkpoint(__lowerCAmelCase							)

   __UpperCamelCase	:			Optional[int]					=					fairseq_model.lm.state_dict()
   __UpperCamelCase	,     __UpperCamelCase	:			Union[str, Any]					=					rename_state_dict(
       __lowerCAmelCase					,							hidden_size=decoder_config.hidden_size							)

   __UpperCamelCase	:			List[Any]					=					TaEncoderModel.from_pretrained("""t5-base"""							)
   __UpperCamelCase	:			Dict					=					EncodecModel.from_pretrained("""facebook/encodec_32khz"""							)
   __UpperCamelCase	:			List[str]					=					MusicgenForCausalLM(__lowerCAmelCase							).eval()

   # load all decoder weights - expect that we'll be missing embeddings and enc-dec projection
   __UpperCamelCase	,     __UpperCamelCase	:			Tuple					=					decoder.load_state_dict(__lowerCAmelCase					,							strict=__lowerCAmelCase							)

   for key in missing_keys.copy():
      if key.startswith(("""text_encoder""", """audio_encoder""")							) or key in EXPECTED_MISSING_KEYS:
         missing_keys.remove(__lowerCAmelCase							)

   if len(__lowerCAmelCase							) > 0:
      raise ValueError(f'Missing key(s) in state_dict: {missing_keys}'							)

   if len(__lowerCAmelCase							) > 0:
      raise ValueError(f'Unexpected key(s) in state_dict: {unexpected_keys}'							)

   # init the composite model
   __UpperCamelCase	:			Tuple					=					MusicgenForConditionalGeneration(text_encoder=__lowerCAmelCase					,							audio_encoder=__lowerCAmelCase					,							decoder=__lowerCAmelCase							)

   # load the pre-trained enc-dec projection (from the decoder state dict)
   model.enc_to_dec_proj.load_state_dict(__lowerCAmelCase							)

   # check we can do a forward pass
   __UpperCamelCase	:			int					=					torch.arange(0					,							8					,							dtype=torch.long							).reshape(2					,							-1							)
   __UpperCamelCase	:			List[Any]					=					input_ids.reshape(2 * 4					,							-1							)

   with torch.no_grad():
      __UpperCamelCase	:			int					=					model(input_ids=__lowerCAmelCase					,							decoder_input_ids=__lowerCAmelCase							).logits

   if logits.shape != (8, 1, 2048):
      raise ValueError("""Incorrect shape for logits"""							)

   # now construct the processor
   __UpperCamelCase	:			Union[str, Any]					=					AutoTokenizer.from_pretrained("""t5-base"""							)
   __UpperCamelCase	:			List[Any]					=					AutoFeatureExtractor.from_pretrained("""facebook/encodec_32khz"""					,							padding_side="""left"""							)

   __UpperCamelCase	:			Any					=					MusicgenProcessor(feature_extractor=__lowerCAmelCase					,							tokenizer=__lowerCAmelCase							)

   # set the appropriate bos/pad token ids
   __UpperCamelCase	:			Tuple					=					2048
   __UpperCamelCase	:			int					=					2048

   # set other default generation config params
   __UpperCamelCase	:			str					=					int(30 * audio_encoder.config.frame_rate							)
   __UpperCamelCase	:			List[str]					=					True
   __UpperCamelCase	:			Tuple					=					3.0

   if pytorch_dump_folder is not None:
      Path(__lowerCAmelCase							).mkdir(exist_ok=__lowerCAmelCase							)
      logger.info(f'Saving model {checkpoint} to {pytorch_dump_folder}'							)
      model.save_pretrained(__lowerCAmelCase							)
      processor.save_pretrained(__lowerCAmelCase							)

   if repo_id:
      logger.info(f'Pushing model {checkpoint} to {repo_id}'							)
      model.push_to_hub(__lowerCAmelCase							)
      processor.push_to_hub(__lowerCAmelCase							)


if __name__ == "__main__":
       UpperCamelCase    							=      argparse.ArgumentParser()
       # Required parameters
       parser.add_argument(
           '--checkpoint',
           default='small',
           type=str,
           help='Checkpoint size of the MusicGen model you\'d like to convert. Can be one of: `[\'small\', \'medium\', \'large\']`.',
       )
       parser.add_argument(
           '--pytorch_dump_folder',
           required=True,
           default=None,
           type=str,
           help='Path to the output PyTorch model directory.',
       )
       parser.add_argument(
           '--push_to_hub', default=None, type=str, help='Where to upload the converted model on the 🤗 hub.'
       )
       parser.add_argument(
           '--device', default='cpu', type=str, help='Torch device to run the conversion, either cpu or cuda.'
       )

       UpperCamelCase    							=      parser.parse_args()
       convert_musicgen_checkpoint(args.checkpoint, args.pytorch_dump_folder, args.push_to_hub)



