





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()
_a									=				logging.get_logger(__name__)

_a									=				{
    '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',
}
_a									=				[
    'lm_head',
]





def lowerCAmelCase__(    __snake_case	,__snake_case	,__snake_case	,__snake_case	,__snake_case							)    ->				Optional[int]:







	'''simple docstring'''







	for attribute in key.split('''.'''							):
		lowerCamelCase__        =		getattr(lowercase_	,lowercase_							)

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

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





def lowerCAmelCase__(    __snake_case	,__snake_case	,__snake_case							)    ->				str:







	'''simple docstring'''







	lowerCamelCase__        =		[]
	lowerCamelCase__        =		fairseq_model.state_dict()

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

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

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

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





def lowerCAmelCase__(    __snake_case	,__snake_case							)    ->				List[str]:







	'''simple docstring'''







	lowerCamelCase__        =		path.split('''.'''							)
	return reduce(lowercase_	,lowercase_	,lowercase_							)





def lowerCAmelCase__(    __snake_case	,__snake_case	,__snake_case	,__snake_case							)    ->				Optional[Any]:







	'''simple docstring'''







	lowerCamelCase__        =		access_by_string(lowercase_	,lowercase_							)
	lowerCamelCase__        =		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.'							)
	lowerCamelCase__        =		fsq_value
	logger.info(F'{full_name} was correctly initialized from {hf_weight_path}.'							)





def lowerCAmelCase__(    __snake_case	,__snake_case	,__snake_case	,__snake_case							)    ->				Optional[Any]:







	'''simple docstring'''







	lowerCamelCase__        =		full_name.split('''conv_layers.'''							)[-1]
	lowerCamelCase__        =		name.split('''.'''							)
	lowerCamelCase__        =		int(items[0]							)
	lowerCamelCase__        =		int(items[1]							)

	lowerCamelCase__        =		name.split('''.'''							)[-1]
	if type_id == 0:
		lowerCamelCase__        =		'''conv'''
	elif type_id == 2:
		lowerCamelCase__        =		'''layer_norm'''
	else:
		unused_weights.append(lowercase_							)
		return

	set_weights(lowercase_	,lowercase_	,lowercase_	,F'conv_layers.{layer_id}.{layer_type}.{weight_type}'							)





def lowerCAmelCase__(    __snake_case	,__snake_case	,__snake_case	,__snake_case							)    ->				Dict:







	'''simple docstring'''







	lowerCamelCase__        =		full_name.split('''pos_conv.'''							)[-1]
	lowerCamelCase__        =		name.split('''.'''							)
	lowerCamelCase__        =		int(items[0]							)
	lowerCamelCase__        =		int(items[1]							)

	lowerCamelCase__        =		name.split('''.'''							)[-1]
	if type_id != 0:
		unused_weights.append(lowercase_							)
		return
	else:
		lowerCamelCase__        =		'''conv'''

	set_weights(lowercase_	,lowercase_	,lowercase_	,F'layers.{layer_id}.{layer_type}.{weight_type}'							)





@torch.no_grad()
def lowerCAmelCase__(    __snake_case	,__snake_case	,__snake_case=None	,__snake_case=None	,__snake_case=True							)    ->				List[Any]:







	'''simple docstring'''







	if config_path is not None:
		lowerCamelCase__        =		DataaVecAudioConfig.from_pretrained(lowercase_							)
	else:
		lowerCamelCase__        =		DataaVecAudioConfig()

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

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

	def load_dataavec(__snake_case							):
		lowerCamelCase__    ,     lowerCamelCase__    ,     lowerCamelCase__        =		fairseq.checkpoint_utils.load_model_ensemble_and_task([path]							)
		return model[0].eval()

	lowerCamelCase__        =		load_dataavec(lowercase_							)

	recursively_load_weights(lowercase_	,lowercase_	,not is_finetuned							)

	lowerCamelCase__        =		WavaVecaProcessor.from_pretrained('''facebook/wav2vec2-large-lv60'''							)

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

	lowerCamelCase__        =		processor(lowercase_	,return_tensors='''pt'''	,padding=lowercase_							)

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

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

		lowerCamelCase__        =		torch.argmax(lowercase_	,dim=-1							)
		lowerCamelCase__        =		processor.batch_decode(lowercase_							)

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

	print(our_output.shape	,their_output.shape							)
	lowerCamelCase__        =		torch.max(torch.abs(our_output - their_output							)							).item()
	print(F'max_absolute_diff = {max_absolute_diff}'							)  # ~ 1e-7
	lowerCamelCase__        =		torch.allclose(lowercase_	,lowercase_	,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(lowercase_							)

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


if __name__ == "__main__":
			_a									=				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"
			)
			_a									=				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 unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin, prepare_image_inputs


if is_torch_available():
			import torch

if is_vision_available():
			from PIL import Image

			from transformers import ViltImageProcessor






class  __A   (   unittest.TestCase			):

			'''simple docstring'''







			def __init__(   self	,  __lowerCAmelCase	,  __lowerCAmelCase=7	,  __lowerCAmelCase=3	,  __lowerCAmelCase=1_8	,  __lowerCAmelCase=3_0	,  __lowerCAmelCase=4_0_0	,  __lowerCAmelCase=True	,  __lowerCAmelCase=None	,  __lowerCAmelCase=2	,  __lowerCAmelCase=True	,  __lowerCAmelCase=[0.5, 0.5, 0.5]	,  __lowerCAmelCase=[0.5, 0.5, 0.5]	,  ):



				'''simple docstring'''







				lowerCamelCase__        =		size if size is not None else {'''shortest_edge''': 3_0}
				lowerCamelCase__        =		parent
				lowerCamelCase__        =		batch_size
				lowerCamelCase__        =		num_channels
				lowerCamelCase__        =		image_size
				lowerCamelCase__        =		min_resolution
				lowerCamelCase__        =		max_resolution
				lowerCamelCase__        =		do_resize
				lowerCamelCase__        =		size
				lowerCamelCase__        =		size_divisor
				lowerCamelCase__        =		do_normalize
				lowerCamelCase__        =		image_mean
				lowerCamelCase__        =		image_std





			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				return {
				    "image_mean": self.image_mean,
				    "image_std": self.image_std,
				    "do_normalize": self.do_normalize,
				    "do_resize": self.do_resize,
				    "size": self.size,
				    "size_divisor": self.size_divisor,
				}





			def 		__lowerCamelCase    (   self	,  __lowerCAmelCase	,  __lowerCAmelCase=False							):



				'''simple docstring'''







				if not batched:
					lowerCamelCase__        =		self.size['''shortest_edge''']
					lowerCamelCase__        =		image_inputs[0]
					if isinstance(__lowerCAmelCase	,  Image.Image							):
						lowerCamelCase__    ,     lowerCamelCase__        =		image.size
					else:
						lowerCamelCase__    ,     lowerCamelCase__        =		image.shape[1], image.shape[2]
					lowerCamelCase__        =		size / min(__lowerCAmelCase	,  __lowerCAmelCase							)
					if h < w:
						lowerCamelCase__    ,     lowerCamelCase__        =		size, scale * w
					else:
						lowerCamelCase__    ,     lowerCamelCase__        =		scale * h, size

					lowerCamelCase__        =		int((1_3_3_3 / 8_0_0) * size							)
					if max(__lowerCAmelCase	,  __lowerCAmelCase							) > max_size:
						lowerCamelCase__        =		max_size / max(__lowerCAmelCase	,  __lowerCAmelCase							)
						lowerCamelCase__        =		newh * scale
						lowerCamelCase__        =		neww * scale

					lowerCamelCase__    ,     lowerCamelCase__        =		int(newh + 0.5							), int(neww + 0.5							)
					lowerCamelCase__    ,     lowerCamelCase__        =		(
					    newh // self.size_divisor * self.size_divisor,
					    neww // self.size_divisor * self.size_divisor,
					)

				else:
					lowerCamelCase__        =		[]
					for image in image_inputs:
						lowerCamelCase__    ,     lowerCamelCase__        =		self.get_expected_values([image]							)
						expected_values.append((expected_height, expected_width)							)
					lowerCamelCase__        =		max(__lowerCAmelCase	,  key=lambda __lowerCAmelCase							: item[0]							)[0]
					lowerCamelCase__        =		max(__lowerCAmelCase	,  key=lambda __lowerCAmelCase							: item[1]							)[1]

				return expected_height, expected_width






@require_torch
@require_vision
class  __A   (   lowerCAmelCase      ,    unittest.TestCase			):

			'''simple docstring'''







			lowerCAmelCase_												=						ViltImageProcessor if is_vision_available() else None





			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				lowerCamelCase__        =		ViltImageProcessingTester(self							)





			@property
			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				return self.image_processor_tester.prepare_image_processor_dict()





			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				lowerCamelCase__        =		self.image_processing_class(**self.image_processor_dict							)
				self.assertTrue(hasattr(__lowerCAmelCase	,  '''image_mean'''							)							)
				self.assertTrue(hasattr(__lowerCAmelCase	,  '''image_std'''							)							)
				self.assertTrue(hasattr(__lowerCAmelCase	,  '''do_normalize'''							)							)
				self.assertTrue(hasattr(__lowerCAmelCase	,  '''do_resize'''							)							)
				self.assertTrue(hasattr(__lowerCAmelCase	,  '''size'''							)							)
				self.assertTrue(hasattr(__lowerCAmelCase	,  '''size_divisor'''							)							)





			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				lowerCamelCase__        =		self.image_processing_class.from_dict(self.image_processor_dict							)
				self.assertEqual(image_processor.size	,  {'''shortest_edge''': 3_0}							)

				lowerCamelCase__        =		self.image_processing_class.from_dict(self.image_processor_dict	,  size=4_2							)
				self.assertEqual(image_processor.size	,  {'''shortest_edge''': 4_2}							)





			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				pass





			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				lowerCamelCase__        =		self.image_processing_class(**self.image_processor_dict							)
				# create random PIL images
				lowerCamelCase__        =		prepare_image_inputs(self.image_processor_tester	,  equal_resolution=__lowerCAmelCase							)
				for image in image_inputs:
					self.assertIsInstance(__lowerCAmelCase	,  Image.Image							)

				# Test not batched input
				lowerCamelCase__        =		image_processing(image_inputs[0]	,  return_tensors='''pt'''							).pixel_values

				lowerCamelCase__    ,     lowerCamelCase__        =		self.image_processor_tester.get_expected_values(__lowerCAmelCase							)
				self.assertEqual(
				    encoded_images.shape	,  (1, self.image_processor_tester.num_channels, expected_height, expected_width)	,  )

				# Test batched
				lowerCamelCase__        =		image_processing(__lowerCAmelCase	,  return_tensors='''pt'''							).pixel_values

				lowerCamelCase__    ,     lowerCamelCase__        =		self.image_processor_tester.get_expected_values(__lowerCAmelCase	,  batched=__lowerCAmelCase							)
				self.assertEqual(
				    encoded_images.shape	,  (
				        self.image_processor_tester.batch_size,
				        self.image_processor_tester.num_channels,
				        expected_height,
				        expected_width,
				    )	,  )





			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				lowerCamelCase__        =		self.image_processing_class(**self.image_processor_dict							)
				# create random numpy tensors
				lowerCamelCase__        =		prepare_image_inputs(self.image_processor_tester	,  equal_resolution=__lowerCAmelCase	,  numpify=__lowerCAmelCase							)
				for image in image_inputs:
					self.assertIsInstance(__lowerCAmelCase	,  np.ndarray							)

				# Test not batched input
				lowerCamelCase__        =		image_processing(image_inputs[0]	,  return_tensors='''pt'''							).pixel_values

				lowerCamelCase__    ,     lowerCamelCase__        =		self.image_processor_tester.get_expected_values(__lowerCAmelCase							)
				self.assertEqual(
				    encoded_images.shape	,  (1, self.image_processor_tester.num_channels, expected_height, expected_width)	,  )

				# Test batched
				lowerCamelCase__        =		image_processing(__lowerCAmelCase	,  return_tensors='''pt'''							).pixel_values

				lowerCamelCase__    ,     lowerCamelCase__        =		self.image_processor_tester.get_expected_values(__lowerCAmelCase	,  batched=__lowerCAmelCase							)
				self.assertEqual(
				    encoded_images.shape	,  (
				        self.image_processor_tester.batch_size,
				        self.image_processor_tester.num_channels,
				        expected_height,
				        expected_width,
				    )	,  )





			def 		__lowerCamelCase    (   self							):



				'''simple docstring'''







				lowerCamelCase__        =		self.image_processing_class(**self.image_processor_dict							)
				# create random PyTorch tensors
				lowerCamelCase__        =		prepare_image_inputs(self.image_processor_tester	,  equal_resolution=__lowerCAmelCase	,  torchify=__lowerCAmelCase							)
				for image in image_inputs:
					self.assertIsInstance(__lowerCAmelCase	,  torch.Tensor							)

				# Test not batched input
				lowerCamelCase__        =		image_processing(image_inputs[0]	,  return_tensors='''pt'''							).pixel_values

				lowerCamelCase__    ,     lowerCamelCase__        =		self.image_processor_tester.get_expected_values(__lowerCAmelCase							)
				self.assertEqual(
				    encoded_images.shape	,  (1, self.image_processor_tester.num_channels, expected_height, expected_width)	,  )

				# Test batched
				lowerCamelCase__        =		image_processing(__lowerCAmelCase	,  return_tensors='''pt'''							).pixel_values

				lowerCamelCase__    ,     lowerCamelCase__        =		self.image_processor_tester.get_expected_values(__lowerCAmelCase	,  batched=__lowerCAmelCase							)
				self.assertEqual(
				    encoded_images.shape	,  (
				        self.image_processor_tester.batch_size,
				        self.image_processor_tester.num_channels,
				        expected_height,
				        expected_width,
				    )	,  )



