



"""simple docstring"""





import gc
import unittest

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

from diffusers import AutoencoderKL, DDIMScheduler, LDMTextToImagePipeline, UNetaDConditionModel
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    load_numpy,
    nightly,
    require_torch_gpu,
    slow,
    torch_device,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
class    lowercase__     (					SCREAMING_SNAKE_CASE       ,      unittest.TestCase   ):







       '''simple docstring'''


       UpperCamelCase		=  LDMTextToImagePipeline
       UpperCamelCase		=  TEXT_TO_IMAGE_PARAMS - {
           '''negative_prompt''',
           '''negative_prompt_embeds''',
           '''cross_attention_kwargs''',
           '''prompt_embeds''',
       }
       UpperCamelCase		=  PipelineTesterMixin.required_optional_params - {
           '''num_images_per_prompt''',
           '''callback''',
           '''callback_steps''',
       }
       UpperCamelCase		=  TEXT_TO_IMAGE_BATCH_PARAMS
       UpperCamelCase		=  False



       def   lowercase__       (							self    :						Any					)				->		List[str]:


             '''simple docstring'''




             torch.manual_seed(0					)
             UpperCAmelCase_    				=							UNetaDConditionModel(
                 block_out_channels=(32, 64)	,       layers_per_block=2	,       sample_size=32	,       in_channels=4	,       out_channels=4	,       down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")	,       up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")	,       cross_attention_dim=32	,       )
             UpperCAmelCase_    				=							DDIMScheduler(
                 beta_start=0.0_0085	,       beta_end=0.012	,       beta_schedule="scaled_linear"	,       clip_sample=_UpperCAmelCase	,       set_alpha_to_one=_UpperCAmelCase	,       )
             torch.manual_seed(0					)
             UpperCAmelCase_    				=							AutoencoderKL(
                 block_out_channels=(32, 64)	,       in_channels=3	,       out_channels=3	,       down_block_types=("DownEncoderBlock2D", "DownEncoderBlock2D")	,       up_block_types=("UpDecoderBlock2D", "UpDecoderBlock2D")	,       latent_channels=4	,       )
             torch.manual_seed(0					)
             UpperCAmelCase_    				=							CLIPTextConfig(
                 bos_token_id=0	,       eos_token_id=2	,       hidden_size=32	,       intermediate_size=37	,       layer_norm_eps=1e-05	,       num_attention_heads=4	,       num_hidden_layers=5	,       pad_token_id=1	,       vocab_size=1000	,       )
             UpperCAmelCase_    				=							CLIPTextModel(_UpperCAmelCase					)
             UpperCAmelCase_    				=							CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"					)

             UpperCAmelCase_    				=							{
                 "unet": unet,
                 "scheduler": scheduler,
                 "vqvae": vae,
                 "bert": text_encoder,
                 "tokenizer": tokenizer,
             }
             return components



       def   lowercase__       (							self    :						Tuple	,       _UpperCAmelCase    :						str	,       _UpperCAmelCase    :						List[Any]=0					)				->		List[Any]:


             '''simple docstring'''




             if str(_UpperCAmelCase					).startswith("mps"					):
                   UpperCAmelCase_    				=							torch.manual_seed(_UpperCAmelCase					)
             else:
                   UpperCAmelCase_    				=							torch.Generator(device=_UpperCAmelCase					).manual_seed(_UpperCAmelCase					)
             UpperCAmelCase_    				=							{
                 "prompt": "A painting of a squirrel eating a burger",
                 "generator": generator,
                 "num_inference_steps": 2,
                 "guidance_scale": 6.0,
                 "output_type": "numpy",
             }
             return inputs







       def   lowercase__       (							self    :						str					)				->		Dict:


             '''simple docstring'''




             UpperCAmelCase_    				=							"cpu"  # ensure determinism for the device-dependent torch.Generator

             UpperCAmelCase_    				=							self.get_dummy_components()
             UpperCAmelCase_    				=							LDMTextToImagePipeline(**_UpperCAmelCase					)
             pipe.to(_UpperCAmelCase					)
             pipe.set_progress_bar_config(disable=_UpperCAmelCase					)

             UpperCAmelCase_    				=							self.get_dummy_inputs(_UpperCAmelCase					)
             UpperCAmelCase_    				=							pipe(**_UpperCAmelCase					).images
             UpperCAmelCase_    				=							image[0, -3:, -3:, -1]

             assert image.shape == (1, 16, 16, 3)
             UpperCAmelCase_    				=							np.array([0.6101, 0.6156, 0.5622, 0.4895, 0.6661, 0.3804, 0.5748, 0.6136, 0.5014]					)

             assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-3



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







       '''simple docstring'''


       def   lowercase__       (							self    :						Optional[Any]					)				->		List[Any]:


             '''simple docstring'''




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



       def   lowercase__       (							self    :						Dict	,       _UpperCAmelCase    :						Optional[int]	,       _UpperCAmelCase    :						str=torch.floataa	,       _UpperCAmelCase    :						str=0					)				->		Any:


             '''simple docstring'''




             UpperCAmelCase_    				=							torch.manual_seed(_UpperCAmelCase					)
             UpperCAmelCase_    				=							np.random.RandomState(_UpperCAmelCase					).standard_normal((1, 4, 32, 32)					)
             UpperCAmelCase_    				=							torch.from_numpy(_UpperCAmelCase					).to(device=_UpperCAmelCase	,       dtype=_UpperCAmelCase					)
             UpperCAmelCase_    				=							{
                 "prompt": "A painting of a squirrel eating a burger",
                 "latents": latents,
                 "generator": generator,
                 "num_inference_steps": 3,
                 "guidance_scale": 6.0,
                 "output_type": "numpy",
             }
             return inputs







       def   lowercase__       (							self    :						Tuple					)				->		Optional[Any]:


             '''simple docstring'''




             UpperCAmelCase_    				=							LDMTextToImagePipeline.from_pretrained("CompVis/ldm-text2im-large-256"					).to(_UpperCAmelCase					)
             pipe.set_progress_bar_config(disable=_UpperCAmelCase					)

             UpperCAmelCase_    				=							self.get_inputs(_UpperCAmelCase					)
             UpperCAmelCase_    				=							pipe(**_UpperCAmelCase					).images
             UpperCAmelCase_    				=							image[0, -3:, -3:, -1].flatten()

             assert image.shape == (1, 256, 256, 3)
             UpperCAmelCase_    				=							np.array([0.5_1825, 0.5_2850, 0.5_2543, 0.5_4258, 0.5_2304, 0.5_2569, 0.5_4363, 0.5_5276, 0.5_6878]					)
             UpperCAmelCase_    				=							np.abs(expected_slice - image_slice					).max()
             assert max_diff < 1e-3



@nightly
@require_torch_gpu
class    lowercase__     (					unittest.TestCase   ):







       '''simple docstring'''


       def   lowercase__       (							self    :						Dict					)				->		str:


             '''simple docstring'''




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



       def   lowercase__       (							self    :						Tuple	,       _UpperCAmelCase    :						List[str]	,       _UpperCAmelCase    :						List[str]=torch.floataa	,       _UpperCAmelCase    :						Dict=0					)				->		Dict:


             '''simple docstring'''




             UpperCAmelCase_    				=							torch.manual_seed(_UpperCAmelCase					)
             UpperCAmelCase_    				=							np.random.RandomState(_UpperCAmelCase					).standard_normal((1, 4, 32, 32)					)
             UpperCAmelCase_    				=							torch.from_numpy(_UpperCAmelCase					).to(device=_UpperCAmelCase	,       dtype=_UpperCAmelCase					)
             UpperCAmelCase_    				=							{
                 "prompt": "A painting of a squirrel eating a burger",
                 "latents": latents,
                 "generator": generator,
                 "num_inference_steps": 50,
                 "guidance_scale": 6.0,
                 "output_type": "numpy",
             }
             return inputs







       def   lowercase__       (							self    :						List[str]					)				->		Union[str, Any]:


             '''simple docstring'''




             UpperCAmelCase_    				=							LDMTextToImagePipeline.from_pretrained("CompVis/ldm-text2im-large-256"					).to(_UpperCAmelCase					)
             pipe.set_progress_bar_config(disable=_UpperCAmelCase					)

             UpperCAmelCase_    				=							self.get_inputs(_UpperCAmelCase					)
             UpperCAmelCase_    				=							pipe(**_UpperCAmelCase					).images[0]

             UpperCAmelCase_    				=							load_numpy(
                 "https://huggingface.co/datasets/diffusers/test-arrays/resolve/main/ldm_text2img/ldm_large_256_ddim.npy"					)
             UpperCAmelCase_    				=							np.abs(expected_image - image					).max()
             assert max_diff < 1e-3









"""simple docstring"""





import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConformerConfig,
    WavaVecaConformerForCTC,
    WavaVecaConformerForPreTraining,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


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

lowerCamelCase      =		{
    """post_extract_proj""": """feature_projection.projection""",
    """encoder.pos_conv.0""": """encoder.pos_conv_embed.conv""",
    """self_attn.linear_k""": """encoder.layers.*.self_attn.linear_k""",
    """self_attn.linear_v""": """encoder.layers.*.self_attn.linear_v""",
    """self_attn.linear_q""": """encoder.layers.*.self_attn.linear_q""",
    """self_attn.pos_bias_u""": """encoder.layers.*.self_attn.pos_bias_u""",
    """self_attn.pos_bias_v""": """encoder.layers.*.self_attn.pos_bias_v""",
    """self_attn.linear_out""": """encoder.layers.*.self_attn.linear_out""",
    """self_attn.linear_pos""": """encoder.layers.*.self_attn.linear_pos""",
    """self_attn.rotary_emb""": """encoder.embed_positions""",
    """self_attn_layer_norm""": """encoder.layers.*.self_attn_layer_norm""",
    """conv_module.pointwise_conv1""": """encoder.layers.*.conv_module.pointwise_conv1""",
    """conv_module.pointwise_conv2""": """encoder.layers.*.conv_module.pointwise_conv2""",
    """conv_module.depthwise_conv""": """encoder.layers.*.conv_module.depthwise_conv""",
    """conv_module.batch_norm""": """encoder.layers.*.conv_module.batch_norm""",
    """conv_module.layer_norm""": """encoder.layers.*.conv_module.layer_norm""",
    """ffn1.w_1""": """encoder.layers.*.ffn1.intermediate_dense""",
    """ffn1.w_2""": """encoder.layers.*.ffn1.output_dense""",
    """ffn1.layer_norm""": """encoder.layers.*.ffn1_layer_norm""",
    """ffn2.w_1""": """encoder.layers.*.ffn2.intermediate_dense""",
    """ffn2.w_2""": """encoder.layers.*.ffn2.output_dense""",
    """ffn2.layer_norm""": """encoder.layers.*.ffn2_layer_norm""",
    """final_layer_norm""": """encoder.layers.*.final_layer_norm""",
    """encoder.layer_norm""": """encoder.layer_norm""",
    """w2v_model.layer_norm""": """feature_projection.layer_norm""",
    """quantizer.weight_proj""": """quantizer.weight_proj""",
    """quantizer.vars""": """quantizer.codevectors""",
    """project_q""": """project_q""",
    """final_proj""": """project_hid""",
    """w2v_encoder.proj""": """lm_head""",
    """mask_emb""": """masked_spec_embed""",
}
lowerCamelCase      =		[
    """lm_head""",
    """quantizer.weight_proj""",
    """quantizer.codevectors""",
    """project_q""",
    """project_hid""",
]
def      a__			(   lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__   ):
      for attribute in key.split("."   ):
            UpperCAmelCase_    				=							getattr(lowerCAmelCase__			,    lowerCAmelCase__   )

      if weight_type is not None:
            UpperCAmelCase_    				=							getattr(lowerCAmelCase__			,    lowerCAmelCase__   ).shape
      else:
            UpperCAmelCase_    				=							hf_pointer.shape

      if hf_shape != value.shape:
            raise ValueError(
                f"""Shape of hf {key + '.' + weight_type if weight_type is not None else ''} is {hf_shape}, but should be"""
                f""" {value.shape} for {full_name}"""   )

      if weight_type == "weight":
            UpperCAmelCase_    				=							value
      elif weight_type == "weight_g":
            UpperCAmelCase_    				=							value
      elif weight_type == "weight_v":
            UpperCAmelCase_    				=							value
      elif weight_type == "bias":
            UpperCAmelCase_    				=							value
      elif weight_type == "running_mean":
            UpperCAmelCase_    				=							value
      elif weight_type == "running_var":
            UpperCAmelCase_    				=							value
      elif weight_type == "num_batches_tracked":
            UpperCAmelCase_    				=							value
      elif weight_type == "inv_freq":
            UpperCAmelCase_    				=							value
      else:
            UpperCAmelCase_    				=							value

      logger.info(f"""{key + '.' + weight_type if weight_type is not None else ''} was initialized from {full_name}."""   )
def      a__			(   lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__   ):
      UpperCAmelCase_    				=							[]
      UpperCAmelCase_    				=							fairseq_model.state_dict()

      UpperCAmelCase_    				=							hf_model.wavaveca_conformer.feature_extractor

      for name, value in fairseq_dict.items():
            UpperCAmelCase_    				=							False
            if "conv_layers" in name:
                  load_conv_layer(
                      lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    hf_model.config.feat_extract_norm == "group"			,    )
                  UpperCAmelCase_    				=							True
            else:
                  for key, mapped_key in MAPPING.items():
                        UpperCAmelCase_    				=							"wav2vec2_conformer." + mapped_key if mapped_key not in TOP_LEVEL_KEYS else mapped_key
                        if key in name or key.split("w2v_model."   )[-1] == name.split("."   )[0]:
                              UpperCAmelCase_    				=							True
                              if "*" in mapped_key:
                                    UpperCAmelCase_    				=							name.split(lowerCAmelCase__   )[0].split("."   )[-2]
                                    UpperCAmelCase_    				=							mapped_key.replace("*"			,    lowerCAmelCase__   )
                              if "pos_bias_u" in name:
                                    UpperCAmelCase_    				=							None
                              elif "pos_bias_v" in name:
                                    UpperCAmelCase_    				=							None
                              elif "weight_g" in name:
                                    UpperCAmelCase_    				=							"weight_g"
                              elif "weight_v" in name:
                                    UpperCAmelCase_    				=							"weight_v"
                              elif "bias" in name:
                                    UpperCAmelCase_    				=							"bias"
                              elif "weight" in name:
                                    # TODO: don't match quantizer.weight_proj
                                    UpperCAmelCase_    				=							"weight"
                              elif "running_mean" in name:
                                    UpperCAmelCase_    				=							"running_mean"
                              elif "inv_freq" in name:
                                    UpperCAmelCase_    				=							"inv_freq"
                              elif "running_var" in name:
                                    UpperCAmelCase_    				=							"running_var"
                              elif "num_batches_tracked" in name:
                                    UpperCAmelCase_    				=							"num_batches_tracked"
                              else:
                                    UpperCAmelCase_    				=							None
                              set_recursively(lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__   )
                        continue
            if not is_used:
                  unused_weights.append(lowerCAmelCase__   )

      logger.warning(f"""Unused weights: {unused_weights}"""   )
def      a__			(   lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__   ):
      UpperCAmelCase_    				=							full_name.split("conv_layers."   )[-1]
      UpperCAmelCase_    				=							name.split("."   )
      UpperCAmelCase_    				=							int(items[0]   )
      UpperCAmelCase_    				=							int(items[1]   )

      if type_id == 0:
            if "bias" in name:
                  if value.shape != feature_extractor.conv_layers[layer_id].conv.bias.data.shape:
                        raise ValueError(
                            f"""{full_name} has size {value.shape}, but"""
                            f""" {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found."""   )
                  UpperCAmelCase_    				=							value
                  logger.info(f"""Feat extract conv layer {layer_id} was initialized from {full_name}."""   )
            elif "weight" in name:
                  if value.shape != feature_extractor.conv_layers[layer_id].conv.weight.data.shape:
                        raise ValueError(
                            f"""{full_name} has size {value.shape}, but"""
                            f""" {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found."""   )
                  UpperCAmelCase_    				=							value
                  logger.info(f"""Feat extract conv layer {layer_id} was initialized from {full_name}."""   )
      elif (type_id == 2 and not use_group_norm) or (type_id == 2 and layer_id == 0 and use_group_norm):
            if "bias" in name:
                  if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape:
                        raise ValueError(
                            f"""{full_name} has size {value.shape}, but"""
                            f""" {feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape} was found."""   )
                  UpperCAmelCase_    				=							value
                  logger.info(f"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""   )
            elif "weight" in name:
                  if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape:
                        raise ValueError(
                            f"""{full_name} has size {value.shape}, but"""
                            f""" {feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape} was found."""   )
                  UpperCAmelCase_    				=							value
                  logger.info(f"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""   )
      else:
            unused_weights.append(lowerCAmelCase__   )
@torch.no_grad()
def      a__			(   lowerCAmelCase__			,    lowerCAmelCase__			,    lowerCAmelCase__=None			,    lowerCAmelCase__=None			,    lowerCAmelCase__=True   ):
      if config_path is not None:
            UpperCAmelCase_    				=							WavaVecaConformerConfig.from_pretrained(lowerCAmelCase__			,    hidden_act="swish"   )
      else:
            UpperCAmelCase_    				=							WavaVecaConformerConfig()

      if "rope" in checkpoint_path:
            UpperCAmelCase_    				=							"rotary"

      if is_finetuned:
            if dict_path:
                  UpperCAmelCase_    				=							Dictionary.load(lowerCAmelCase__   )

                  # important change bos & pad token id since CTC symbol is <pad> and
                  # not <s> as in fairseq
                  UpperCAmelCase_    				=							target_dict.pad_index
                  UpperCAmelCase_    				=							target_dict.bos_index
                  UpperCAmelCase_    				=							target_dict.eos_index
                  UpperCAmelCase_    				=							len(target_dict.symbols   )
                  UpperCAmelCase_    				=							os.path.join(lowerCAmelCase__			,    "vocab.json"   )
                  if not os.path.isdir(lowerCAmelCase__   ):
                        logger.error("--pytorch_dump_folder_path ({}) should be a directory".format(lowerCAmelCase__   )   )
                        return
                  os.makedirs(lowerCAmelCase__			,    exist_ok=lowerCAmelCase__   )
                  UpperCAmelCase_    				=							target_dict.indices

                  # fairseq has the <pad> and <s> switched
                  UpperCAmelCase_    				=							0
                  UpperCAmelCase_    				=							1
                  with open(lowerCAmelCase__			,    "w"			,    encoding="utf-8"   ) as vocab_handle:
                        json.dump(lowerCAmelCase__			,    lowerCAmelCase__   )
                  UpperCAmelCase_    				=							WavaVecaCTCTokenizer(
                      lowerCAmelCase__			,    unk_token=target_dict.unk_word			,    pad_token=target_dict.pad_word			,    bos_token=target_dict.bos_word			,    eos_token=target_dict.eos_word			,    word_delimiter_token="|"			,    do_lower_case=lowerCAmelCase__			,    )
                  UpperCAmelCase_    				=							True if config.feat_extract_norm == "layer" else False
                  UpperCAmelCase_    				=							WavaVecaFeatureExtractor(
                      feature_size=1			,    sampling_rate=16000			,    padding_value=0			,    do_normalize=lowerCAmelCase__			,    return_attention_mask=lowerCAmelCase__			,    )
                  UpperCAmelCase_    				=							WavaVecaProcessor(feature_extractor=lowerCAmelCase__			,    tokenizer=lowerCAmelCase__   )
                  processor.save_pretrained(lowerCAmelCase__   )

            UpperCAmelCase_    				=							WavaVecaConformerForCTC(lowerCAmelCase__   )
      else:
            UpperCAmelCase_    				=							WavaVecaConformerForPreTraining(lowerCAmelCase__   )

      if is_finetuned:
            UpperCAmelCase_				,					UpperCAmelCase_				,					UpperCAmelCase_    				=							fairseq.checkpoint_utils.load_model_ensemble_and_task(
                [checkpoint_path]			,    arg_overrides={"data": "/".join(dict_path.split("/"   )[:-1]   )}   )
      else:
            UpperCAmelCase_    				=							argparse.Namespace(task="audio_pretraining"   )
            UpperCAmelCase_    				=							fairseq.tasks.setup_task(lowerCAmelCase__   )

            UpperCAmelCase_				,					UpperCAmelCase_				,					UpperCAmelCase_    				=							fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]			,    task=lowerCAmelCase__   )

      UpperCAmelCase_    				=							model[0].eval()

      recursively_load_weights(lowerCAmelCase__			,    lowerCAmelCase__			,    not is_finetuned   )

      hf_wavavec.save_pretrained(lowerCAmelCase__   )


if __name__ == "__main__":
  lowerCamelCase      =		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 fairseq checkpoint""")
  parser.add_argument("""--dict_path""", default=None, type=str, help="""Path to dict of fine-tuned model""")
  parser.add_argument("""--config_path""", default=None, type=str, help="""Path to hf config.json of model to convert""")
  parser.add_argument(
      """--not_finetuned""", action="""store_true""", help="""Whether the model to convert is a fine-tuned model or not"""
  )
  lowerCamelCase      =		parser.parse_args()
  convert_wavaveca_conformer_checkpoint(
      args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
  )




