




from dataclasses import dataclass
from typing import Any, Dict, Optional

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..models.embeddings import ImagePositionalEmbeddings
from ..utils import BaseOutput, deprecate
from .attention import BasicTransformerBlock
from .embeddings import PatchEmbed
from .modeling_utils import ModelMixin




@dataclass
class    snake_case__							(_UpperCamelCase       ):
				"""simple docstring"""



				SCREAMING_SNAKE_CASE_	:							torch.FloatTensor




class    snake_case__							(_UpperCamelCase   ,	_UpperCamelCase       ):
				"""simple docstring"""



				@register_to_config
				def __init__(				self    :					List[Any] ,				__lowerCamelCase    :					int = 16 ,				__lowerCamelCase    :					int = 88 ,				__lowerCamelCase    :					Optional[int] = None ,				__lowerCamelCase    :					Optional[int] = None ,				__lowerCamelCase    :					int = 1 ,				__lowerCamelCase    :					float = 0.0 ,				__lowerCamelCase    :					int = 32 ,				__lowerCamelCase    :					Optional[int] = None ,				__lowerCamelCase    :					bool = False ,				__lowerCamelCase    :					Optional[int] = None ,				__lowerCamelCase    :					Optional[int] = None ,				__lowerCamelCase    :					Optional[int] = None ,				__lowerCamelCase    :					str = "geglu" ,				__lowerCamelCase    :					Optional[int] = None ,				__lowerCamelCase    :					bool = False ,				__lowerCamelCase    :					bool = False ,				__lowerCamelCase    :					bool = False ,				__lowerCamelCase    :					str = "layer_norm" ,				__lowerCamelCase    :					bool = True ,				) ->					Optional[int]:
							super().__init__()
							a   =      use_linear_projection
							a   =      num_attention_heads
							a   =      attention_head_dim
							a   =      num_attention_heads * attention_head_dim

							# 1. Transformer2DModel can process both standard continuous images of shape `(batch_size, num_channels, width, height)` as well as quantized image embeddings of shape `(batch_size, num_image_vectors)`
							# Define whether input is continuous or discrete depending on configuration
							a   =      (in_channels is not None) and (patch_size is None)
							a   =      num_vector_embeds is not None
							a   =      in_channels is not None and patch_size is not None

							if norm_type == "layer_norm" and num_embeds_ada_norm is not None:
										a   =      (
										    f"""The configuration file of this model: {self.__class__} is outdated. `norm_type` is either not set or"""
										    " incorrectly set to `'layer_norm'`.Make sure to set `norm_type` to `'ada_norm'` in the config."
										    " Please make sure to update the config accordingly as leaving `norm_type` might led to incorrect"
										    " results in future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it"
										    " would be very nice if you could open a Pull request for the `transformer/config.json` file"
										)
										deprecate("norm_type!=num_embeds_ada_norm" ,				"1.0.0" ,				__lowerCamelCase ,				standard_warn=__lowerCamelCase							)
										a   =      "ada_norm"

							if self.is_input_continuous and self.is_input_vectorized:
										raise ValueError(
										    f"""Cannot define both `in_channels`: {in_channels} and `num_vector_embeds`: {num_vector_embeds}. Make"""
										    " sure that either `in_channels` or `num_vector_embeds` is None."							)
							elif self.is_input_vectorized and self.is_input_patches:
										raise ValueError(
										    f"""Cannot define both `num_vector_embeds`: {num_vector_embeds} and `patch_size`: {patch_size}. Make"""
										    " sure that either `num_vector_embeds` or `num_patches` is None."							)
							elif not self.is_input_continuous and not self.is_input_vectorized and not self.is_input_patches:
										raise ValueError(
										    f"""Has to define `in_channels`: {in_channels}, `num_vector_embeds`: {num_vector_embeds}, or patch_size:"""
										    f""" {patch_size}. Make sure that `in_channels`, `num_vector_embeds` or `num_patches` is not None."""							)

							# 2. Define input layers
							if self.is_input_continuous:
										a   =      in_channels

										a   =      torch.nn.GroupNorm(num_groups=__lowerCamelCase ,				num_channels=__lowerCamelCase ,				eps=1e-6 ,				affine=__lowerCamelCase							)
										if use_linear_projection:
													a   =      nn.Linear(__lowerCamelCase ,				__lowerCamelCase							)
										else:
													a   =      nn.Convad(__lowerCamelCase ,				__lowerCamelCase ,				kernel_size=1 ,				stride=1 ,				padding=0							)
							elif self.is_input_vectorized:
										assert sample_size is not None, "Transformer2DModel over discrete input must provide sample_size"
										assert num_vector_embeds is not None, "Transformer2DModel over discrete input must provide num_embed"

										a   =      sample_size
										a   =      sample_size
										a   =      num_vector_embeds
										a   =      self.height * self.width

										a   =      ImagePositionalEmbeddings(
										    num_embed=__lowerCamelCase ,				embed_dim=__lowerCamelCase ,				height=self.height ,				width=self.width							)
							elif self.is_input_patches:
										assert sample_size is not None, "Transformer2DModel over patched input must provide sample_size"

										a   =      sample_size
										a   =      sample_size

										a   =      patch_size
										a   =      PatchEmbed(
										    height=__lowerCamelCase ,				width=__lowerCamelCase ,				patch_size=__lowerCamelCase ,				in_channels=__lowerCamelCase ,				embed_dim=__lowerCamelCase ,				)

							# 3. Define transformers blocks
							a   =      nn.ModuleList(
							    [
							        BasicTransformerBlock(
							            __lowerCamelCase ,				__lowerCamelCase ,				__lowerCamelCase ,				dropout=__lowerCamelCase ,				cross_attention_dim=__lowerCamelCase ,				activation_fn=__lowerCamelCase ,				num_embeds_ada_norm=__lowerCamelCase ,				attention_bias=__lowerCamelCase ,				only_cross_attention=__lowerCamelCase ,				upcast_attention=__lowerCamelCase ,				norm_type=__lowerCamelCase ,				norm_elementwise_affine=__lowerCamelCase ,				)
							        for d in range(__lowerCamelCase							)
							    ]							)

							# 4. Define output layers
							a   =      in_channels if out_channels is None else out_channels
							if self.is_input_continuous:
										# TODO: should use out_channels for continuous projections
										if use_linear_projection:
													a   =      nn.Linear(__lowerCamelCase ,				__lowerCamelCase							)
										else:
													a   =      nn.Convad(__lowerCamelCase ,				__lowerCamelCase ,				kernel_size=1 ,				stride=1 ,				padding=0							)
							elif self.is_input_vectorized:
										a   =      nn.LayerNorm(__lowerCamelCase							)
										a   =      nn.Linear(__lowerCamelCase ,				self.num_vector_embeds - 1							)
							elif self.is_input_patches:
										a   =      nn.LayerNorm(__lowerCamelCase ,				elementwise_affine=__lowerCamelCase ,				eps=1e-6							)
										a   =      nn.Linear(__lowerCamelCase ,				2 * inner_dim							)
										a   =      nn.Linear(__lowerCamelCase ,				patch_size * patch_size * self.out_channels							)





				def 							__UpperCAmelCase							(				self    :					Union[str, Any] ,				__lowerCamelCase    :					torch.Tensor ,				__lowerCamelCase    :					Optional[torch.Tensor] = None ,				__lowerCamelCase    :					Optional[torch.LongTensor] = None ,				__lowerCamelCase    :					Optional[torch.LongTensor] = None ,				__lowerCamelCase    :					Dict[str, Any] = None ,				__lowerCamelCase    :					Optional[torch.Tensor] = None ,				__lowerCamelCase    :					Optional[torch.Tensor] = None ,				__lowerCamelCase    :					bool = True ,				) ->					int:
							# ensure attention_mask is a bias, and give it a singleton query_tokens dimension.
							#   we may have done this conversion already, e.g. if we came here via UNet2DConditionModel#forward.
							#   we can tell by counting dims; if ndim == 2: it's a mask rather than a bias.
							# expects mask of shape:
							#   [batch, key_tokens]
							# adds singleton query_tokens dimension:
							#   [batch,                    1, key_tokens]
							# this helps to broadcast it as a bias over attention scores, which will be in one of the following shapes:
							#   [batch,  heads, query_tokens, key_tokens] (e.g. torch sdp attn)
							#   [batch * heads, query_tokens, key_tokens] (e.g. xformers or classic attn)
							if attention_mask is not None and attention_mask.ndim == 2:
										# assume that mask is expressed as:
										#   (1 = keep,      0 = discard)
										# convert mask into a bias that can be added to attention scores:
										#       (keep = +0,     discard = -10000.0)
										a   =      (1 - attention_mask.to(hidden_states.dtype							)) * -10_000.0
										a   =      attention_mask.unsqueeze(1							)

							# convert encoder_attention_mask to a bias the same way we do for attention_mask
							if encoder_attention_mask is not None and encoder_attention_mask.ndim == 2:
										a   =      (1 - encoder_attention_mask.to(hidden_states.dtype							)) * -10_000.0
										a   =      encoder_attention_mask.unsqueeze(1							)

							# 1. Input
							if self.is_input_continuous:
										a   ,		a   ,		a   ,		a   =      hidden_states.shape
										a   =      hidden_states

										a   =      self.norm(__lowerCamelCase							)
										if not self.use_linear_projection:
													a   =      self.proj_in(__lowerCamelCase							)
													a   =      hidden_states.shape[1]
													a   =      hidden_states.permute(0 ,				2 ,				3 ,				1							).reshape(__lowerCamelCase ,				height * width ,				__lowerCamelCase							)
										else:
													a   =      hidden_states.shape[1]
													a   =      hidden_states.permute(0 ,				2 ,				3 ,				1							).reshape(__lowerCamelCase ,				height * width ,				__lowerCamelCase							)
													a   =      self.proj_in(__lowerCamelCase							)
							elif self.is_input_vectorized:
										a   =      self.latent_image_embedding(__lowerCamelCase							)
							elif self.is_input_patches:
										a   =      self.pos_embed(__lowerCamelCase							)

							# 2. Blocks
							for block in self.transformer_blocks:
										a   =      block(
										    __lowerCamelCase ,				attention_mask=__lowerCamelCase ,				encoder_hidden_states=__lowerCamelCase ,				encoder_attention_mask=__lowerCamelCase ,				timestep=__lowerCamelCase ,				cross_attention_kwargs=__lowerCamelCase ,				class_labels=__lowerCamelCase ,				)

							# 3. Output
							if self.is_input_continuous:
										if not self.use_linear_projection:
													a   =      hidden_states.reshape(__lowerCamelCase ,				__lowerCamelCase ,				__lowerCamelCase ,				__lowerCamelCase							).permute(0 ,				3 ,				1 ,				2							).contiguous()
													a   =      self.proj_out(__lowerCamelCase							)
										else:
													a   =      self.proj_out(__lowerCamelCase							)
													a   =      hidden_states.reshape(__lowerCamelCase ,				__lowerCamelCase ,				__lowerCamelCase ,				__lowerCamelCase							).permute(0 ,				3 ,				1 ,				2							).contiguous()

										a   =      hidden_states + residual
							elif self.is_input_vectorized:
										a   =      self.norm_out(__lowerCamelCase							)
										a   =      self.out(__lowerCamelCase							)
										# (batch, self.num_vector_embeds - 1, self.num_latent_pixels)
										a   =      logits.permute(0 ,				2 ,				1							)

										# log(p(x_0))
										a   =      F.log_softmax(logits.double() ,				dim=1							).float()
							elif self.is_input_patches:
										# TODO: cleanup!
										a   =      self.transformer_blocks[0].norma.emb(
										    __lowerCamelCase ,				__lowerCamelCase ,				hidden_dtype=hidden_states.dtype							)
										a   ,		a   =      self.proj_out_a(F.silu(__lowerCamelCase							)							).chunk(2 ,				dim=1							)
										a   =      self.norm_out(__lowerCamelCase							) * (1 + scale[:, None]) + shift[:, None]
										a   =      self.proj_out_a(__lowerCamelCase							)

										# unpatchify
										a   =      a   =      int(hidden_states.shape[1] ** 0.5							)
										a   =      hidden_states.reshape(
										    shape=(-1, height, width, self.patch_size, self.patch_size, self.out_channels)							)
										a   =      torch.einsum("nhwpqc->nchpwq" ,				__lowerCamelCase							)
										a   =      hidden_states.reshape(
										    shape=(-1, self.out_channels, height * self.patch_size, width * self.patch_size)							)

							if not return_dict:
										return (output,)

							return TransformeraDModelOutput(sample=__lowerCamelCase							)












import argparse
import os
from functools import reduce

import fairseq
import torch
from datasets import load_dataset

from transformers import WavaVecaProcessor, logging
from transformers.models.dataavec.configuration_dataavec_audio import DataaVecAudioConfig

# Copied from https://github.com/pytorch/fairseq/blob/main/examples/data2vec/models/data2vec_audio.py
from transformers.models.dataavec.dataavec_audio import DataaVecAudioModel as Dummy  # noqa: F401
from transformers.models.dataavec.modeling_dataavec_audio import DataaVecAudioForCTC, DataaVecAudioModel


logging.set_verbosity_info()
__lowerCAmelCase		:    Dict	      =		logging.get_logger(__name__)

__lowerCAmelCase		:    Tuple	      =		{
    'post_extract_proj': 'feature_projection.projection',
    'models.0.layer_norm': 'feature_projection.layer_norm',
    'self_attn.k_proj': 'encoder.layers.*.attention.k_proj',
    'self_attn.v_proj': 'encoder.layers.*.attention.v_proj',
    'self_attn.q_proj': 'encoder.layers.*.attention.q_proj',
    'self_attn.out_proj': 'encoder.layers.*.attention.out_proj',
    'self_attn_layer_norm': 'encoder.layers.*.layer_norm',
    'fc1': 'encoder.layers.*.feed_forward.intermediate_dense',
    'fc2': 'encoder.layers.*.feed_forward.output_dense',
    'final_layer_norm': 'encoder.layers.*.final_layer_norm',
    'encoder.layer_norm': 'encoder.layer_norm',
    'w2v_model.layer_norm': 'feature_projection.layer_norm',
    'w2v_encoder.proj': 'lm_head',
    'mask_emb': 'masked_spec_embed',
}
__lowerCAmelCase		:    List[Any]	      =		[
    'lm_head',
]




def       __magic_name__    (			A		: int,  A		: List[Any],  A		: Tuple,  A		: Dict,  A		: List[str]       ):






			'''simple docstring'''







			for attribute in key.split("."       ):
						a   =      getattr(A,  A       )

			if weight_type is not None:
						a   =      getattr(A,  A       ).shape
			else:
						a   =      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":
						a   =      value
			elif weight_type == "weight_g":
						a   =      value
			elif weight_type == "weight_v":
						a   =      value
			elif weight_type == "bias":
						a   =      value
			else:
						a   =      value

			logger.info(F"""{key + "." + weight_type if weight_type is not None else ""} was initialized from {full_name}."""       )




def       __magic_name__    (			A		: Optional[Any],  A		: int,  A		: Dict       ):






			'''simple docstring'''







			a   =      []
			a   =      fairseq_model.state_dict()

			if not is_headless:
						a   =      hf_model.dataavec_audio.feature_extractor
						a   =      hf_model.dataavec_audio.encoder.pos_conv_embed

			else:
						a   =      hf_model.feature_extractor
						a   =      hf_model.encoder.pos_conv_embed

			for name, value in fairseq_dict.items():
						a   =      False
						if "conv_layers" in name:
									load_conv_layer(
									    A,  A,  A,  A,  )
									a   =      True
						elif "pos_conv" in name:
									load_pos_conv_layer(
									    A,  A,  A,  A,  )
									a   =      True
						else:
									for key, mapped_key in MAPPING.items():
												if not is_headless:
															a   =      "data2vec_audio." + 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]:
															a   =      True
															if "*" in mapped_key:
																		a   =      name.split(A       )[0].split("."       )[-2]
																		a   =      mapped_key.replace("*",  A       )
															if "weight_g" in name:
																		a   =      "weight_g"
															elif "weight_v" in name:
																		a   =      "weight_v"
															elif "bias" in name:
																		a   =      "bias"
															elif "weight" in name:
																		# TODO: don't match quantizer.weight_proj
																		a   =      "weight"
															else:
																		a   =      None
															set_recursively(A,  A,  A,  A,  A       )
												continue
						if not is_used:
									unused_weights.append(A       )

			logger.warning(F"""Unused weights: {unused_weights}"""       )




def       __magic_name__    (			A		: str,  A		: Dict       ):






			'''simple docstring'''







			a   =      path.split("."       )
			return reduce(A,  A,  A       )




def       __magic_name__    (			A		: Optional[int],  A		: Tuple,  A		: List[str],  A		: Optional[int]       ):






			'''simple docstring'''







			a   =      access_by_string(A,  A       )
			a   =      hf_weight.data

			if fsq_value.shape != hf_value.shape:
						raise ValueError(F"""{full_name} has size {fsq_value.shape}, but {hf_value.shape} was found."""       )
			a   =      fsq_value
			logger.info(F"""{full_name} was correctly initialized from {hf_weight_path}."""       )




def       __magic_name__    (			A		: Optional[Any],  A		: List[str],  A		: Optional[Any],  A		: Optional[int]       ):






			'''simple docstring'''







			a   =      full_name.split("conv_layers."       )[-1]
			a   =      name.split("."       )
			a   =      int(items[0]       )
			a   =      int(items[1]       )

			a   =      name.split("."       )[-1]
			if type_id == 0:
						a   =      "conv"
			elif type_id == 2:
						a   =      "layer_norm"
			else:
						unused_weights.append(A       )
						return

			set_weights(A,  A,  A,  F"""conv_layers.{layer_id}.{layer_type}.{weight_type}"""       )




def       __magic_name__    (			A		: int,  A		: Dict,  A		: List[Any],  A		: Tuple       ):






			'''simple docstring'''







			a   =      full_name.split("pos_conv."       )[-1]
			a   =      name.split("."       )
			a   =      int(items[0]       )
			a   =      int(items[1]       )

			a   =      name.split("."       )[-1]
			if type_id != 0:
						unused_weights.append(A       )
						return
			else:
						a   =      "conv"

			set_weights(A,  A,  A,  F"""layers.{layer_id}.{layer_type}.{weight_type}"""       )




@torch.no_grad()
def       __magic_name__    (			A		: str,  A		: Dict,  A		: int=None,  A		: Tuple=None,  A		: List[str]=True       ):






			'''simple docstring'''







			if config_path is not None:
						a   =      DataaVecAudioConfig.from_pretrained(A       )
			else:
						a   =      DataaVecAudioConfig()

			if not is_finetuned:
						# Modify final_proj layer name
						a   =      DataaVecAudioModel(A       )
						a   =      os.path.dirname(A       )

						a   =      torch.load(A       )
						a   =      state_dict["model"].pop("final_proj.0.weight"       )
						a   =      state_dict["model"].pop("final_proj.0.bias"       )
						a   =      os.path.join(A,  "converted.pt"       )
						torch.save(A,  A       )
			else:
						a   =      DataaVecAudioForCTC(A       )
						a   =      checkpoint_path

			def load_dataavec(A		: Optional[Any]       ):
						a   ,		a   ,		a   =      fairseq.checkpoint_utils.load_model_ensemble_and_task([path]       )
						return model[0].eval()

			a   =      load_dataavec(A       )

			recursively_load_weights(A,  A,  not is_finetuned       )

			a   =      WavaVecaProcessor.from_pretrained("facebook/wav2vec2-large-lv60"       )

			a   =      load_dataset("patrickvonplaten/librispeech_asr_dummy",  "clean",  split="validation"       )
			a   =      [x["array"] for x in ds[:4]["audio"]]

			a   =      processor(A,  return_tensors="pt",  padding=A       )

			a   =      inputs.input_values
			a   =      inputs.attention_mask
			#    input_values = inputs.input_values[:, :-1]
			#    attention_mask = inputs.attention_mask[:, :-1]

			hf_wavavec.eval()
			model.eval()
			if is_finetuned:
						a   =      model(source=A,  padding_mask=(1 - attention_mask),  mask=A,  features_only=A       )[
						    "encoder_out"
						].transpose(0,  1       )
						a   =      hf_wavavec(A,  attention_mask=A       )["logits"]

						a   =      torch.argmax(A,  dim=-1       )
						a   =      processor.batch_decode(A       )

						print(F"""Expected Output: {ds[:4]["text"]}, Pred: {output_string}"""       )
			else:
						a   =      model(source=A,  padding_mask=(1 - attention_mask),  mask=A,  features_only=A       )[
						    "layer_results"
						][-1][0].transpose(0,  1       )
						a   =      hf_wavavec(A,  attention_mask=A       )["last_hidden_state"]

			print(our_output.shape,  their_output.shape       )
			a   =      torch.max(torch.abs(our_output - their_output       )       ).item()
			print(F"""max_absolute_diff = {max_absolute_diff}"""       )  # ~ 1e-7
			a   =      torch.allclose(A,  A,  atol=1E-3       )
			print("Do both models output the same tensors?",  "🔥" if success else "💩"       )
			if not success:
						raise Exception("Something went wRoNg"       )

			hf_wavavec.save_pretrained(A       )

			if is_finetuned:
						processor.save_pretrained(A       )
			else:
						processor.feature_extractor.save_pretrained(A       )


if __name__ == "__main__":
		__lowerCAmelCase		:    List[str]	      =		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		:    int	      =		parser.parse_args()
		convert_wavaveca_checkpoint(
		    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
		)






