



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()
_lowercase:				List[Any]       =	logging.get_logger(__name__)

_lowercase:				Union[str, Any]       =	{
    "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",
}
_lowercase:				Optional[Any]       =	[
    "lm_head",
]






def 					a(   A :					List[str]				,  A :					Any				,  A :					str				,  A :					List[str]				,  A :					Optional[Any] )	-> Tuple:
					"""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 					a(   A :					List[str]				,  A :					List[Any]				,  A :					Optional[Any] )	-> List[Any]:
					"""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 					a(   A :					str				,  A :					Dict )	-> Any:
					"""simple docstring"""





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






def 					a(   A :					Optional[int]				,  A :					Dict				,  A :					Union[str, Any]				,  A :					Tuple )	-> 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 					a(   A :					Optional[int]				,  A :					Tuple				,  A :					Optional[int]				,  A :					Optional[int] )	-> Tuple:
					"""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 					a(   A :					Tuple				,  A :					Optional[int]				,  A :					Optional[Any]				,  A :					str )	-> List[str]:
					"""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 					a(   A :					Dict				,  A :					List[Any]				,  A :					Union[str, Any]=None				,  A :					List[Any]=None				,  A :					Dict=True )	-> Any:
					"""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 :					List[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__":
		_lowercase:				Any       =	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"
		)
		_lowercase:				Optional[Any]       =	parser.parse_args()
		convert_wavaveca_checkpoint(
		    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
		)





import argparse

import torch

from transformers import ChineseCLIPConfig, ChineseCLIPModel






def 					a(   A :					Union[str, Any]				,  A :					List[str]				,  A :					Any )	-> Tuple:
					"""simple docstring"""





					a  ,       a  ,       a        =     pt_weights[f'''{prefix}.in_proj_weight'''].chunk(3				,  dim=0 )
					a  ,       a  ,       a        =     pt_weights[f'''{prefix}.in_proj_bias'''].chunk(3				,  dim=0 )

					a        =     pt_weights[f'''{prefix}.out_proj.weight''']
					a        =     pt_weights[f'''{prefix}.out_proj.bias''']

					a        =     q_proj
					a        =     q_proj_bias

					a        =     k_proj
					a        =     k_proj_bias

					a        =     v_proj
					a        =     v_proj_bias

					a        =     out_proj_weights
					a        =     out_proj_bias






def 					a(   A :					List[str]				,  A :					List[Any]				,  A :					int )	-> List[Any]:
					"""simple docstring"""





					copy_linear(hf_mlp.fca				,  A				,  f'''{prefix}.c_fc''' )
					copy_linear(hf_mlp.fca				,  A				,  f'''{prefix}.c_proj''' )






def 					a(   A :					Optional[Any]				,  A :					List[Any]				,  A :					List[str] )	-> Dict:
					"""simple docstring"""





					a        =     pt_weights[f'''{prefix}.weight'''].data
					a        =     pt_weights[f'''{prefix}.bias'''].data






def 					a(   A :					int				,  A :					Optional[Any]				,  A :					List[Any] )	-> Union[str, Any]:
					"""simple docstring"""





					copy_linear(hf_layer.layer_norma				,  A				,  f'''{prefix}.ln_1''' )
					copy_linear(hf_layer.layer_norma				,  A				,  f'''{prefix}.ln_2''' )

					# copy MLP
					copy_mlp(hf_layer.mlp				,  A				,  f'''{prefix}.mlp''' )

					# copy attn
					copy_attn_layer(hf_layer.self_attn				,  A				,  f'''{prefix}.attn''' )






def 					a(   A :					str				,  A :					Optional[int]				,  A :					List[str] )	-> Dict:
					"""simple docstring"""





					for layer_id, hf_layer in enumerate(A ):
										copy_layer(A				,  A				,  f'''{prefix}.{layer_id}''' )






def 					a(   A :					Dict				,  A :					Union[str, Any] )	-> Optional[Any]:
					"""simple docstring"""





					a        =     pt_weights["text_projection"].data.T

					# copy text encoder
					for name, param in hf_model.text_model.named_parameters():
										a        =     pt_weights[f'''bert.{name}'''].data






def 					a(   A :					Optional[Any]				,  A :					Optional[Any] )	-> List[str]:
					"""simple docstring"""





					a        =     pt_weights["visual.proj"].data.T

					# copy layer norms
					copy_linear(hf_model.vision_model.pre_layrnorm				,  A				,  "visual.ln_pre" )
					copy_linear(hf_model.vision_model.post_layernorm				,  A				,  "visual.ln_post" )

					# copy embeddings
					a        =     pt_weights["visual.conv1.weight"].data
					a        =     pt_weights["visual.class_embedding"].data
					a        =     pt_weights["visual.positional_embedding"].data

					# copy encoder
					copy_layers(hf_model.vision_model.encoder.layers				,  A				,  "visual.transformer.resblocks" )






@torch.no_grad()
def 					a(   A :					str				,  A :					List[str]				,  A :					Union[str, Any]=None )	-> Tuple:
					"""simple docstring"""





					assert config_path is not None, "Please specify the ChineseCLIP model config of the corresponding model size."
					a        =     ChineseCLIPConfig.from_pretrained(A )

					a        =     ChineseCLIPModel(A ).eval()

					a        =     torch.load(A				,  map_location="cpu" )["state_dict"]
					a        =     {(name[7:] if name.startswith("module." ) else name): value for name, value in pt_weights.items()}

					copy_text_model_and_projection(A				,  A )
					copy_vision_model_and_projection(A				,  A )
					a        =     pt_weights["logit_scale"].data

					hf_model.save_pretrained(A )


if __name__ == "__main__":
				_lowercase:				List[Any]       =	argparse.ArgumentParser()
				parser.add_argument(
				    "--pytorch_dump_folder_path",
				    default=None,
				    type=str,
				    help="Path to the output folder storing converted hf PyTorch model.",
				)
				parser.add_argument(
				    "--checkpoint_path", default=None, type=str, help="Path to original github format ChineseCLIP checkpoint."
				)
				parser.add_argument(
				    "--config_path", default=None, required=True, type=str, help="Path to hf config.json of model to convert."
				)
				_lowercase:				int       =	parser.parse_args()

				convert_chinese_clip_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path)
				print("The conversion is finished!")
