





"""simple docstring"""



import argparse

import torch
from saprl.hub import distilhubert

from transformers import HubertConfig, HubertModel, WavaVecaFeatureExtractor, logging


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

a_				=    {
    """post_extract_proj""": """feature_projection.projection""",
    """encoder.pos_conv.0""": """encoder.pos_conv_embed.conv""",
    """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""",
    """mask_emb""": """masked_spec_embed""",
}







def 	__lowercase  (     snake_case_						:						List[str]       ,snake_case_						:						Tuple       ,snake_case_						:						Optional[int]       ,snake_case_						:						Tuple       ,snake_case_						:						List[str]		)     ->Any:



				'''simple docstring'''







				for attribute in key.split('''.'''		):
								__A					:		str						=  getattr(snake_case_       ,snake_case_		)

				if weight_type is not None:
								__A					:		Tuple						=  getattr(snake_case_       ,snake_case_		).shape
				else:
								__A					:		str						=  hf_pointer.shape

				assert hf_shape == value.shape, (
				    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					:		str						=  value
				elif weight_type == "weight_g":
								__A					:		List[str]						=  value
				elif weight_type == "weight_v":
								__A					:		str						=  value
				elif weight_type == "bias":
								__A					:		Optional[int]						=  value
				else:
								__A					:		str						=  value

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







def 	__lowercase  (     snake_case_						:						Union[str, Any]       ,snake_case_						:						Optional[Any]		)     ->Dict:



				'''simple docstring'''







				__A					:		str						=  []
				__A					:		Union[str, Any]						=  fairseq_model.state_dict()

				__A					:		Dict						=  hf_model.feature_extractor

				for name, value in fairseq_dict.items():
								__A					:		Tuple						=  False
								if "conv_layers" in name:
												load_conv_layer(
												    snake_case_       ,snake_case_       ,snake_case_       ,snake_case_       ,hf_model.config.feat_extract_norm == '''group'''       ,)
												__A					:		Any						=  True
								else:
												for key, mapped_key in MAPPING.items():
																__A					:		List[str]						=  mapped_key

																if key in name:
																				__A					:		Any						=  True
																				if "*" in mapped_key:
																								__A					:		List[Any]						=  name.split(snake_case_		)[0].split('''.'''		)[-2]
																								__A					:		Union[str, Any]						=  mapped_key.replace('''*'''       ,snake_case_		)
																				if "weight_g" in name:
																								__A					:		Tuple						=  '''weight_g'''
																				elif "weight_v" in name:
																								__A					:		int						=  '''weight_v'''
																				elif "weight" in name:
																								__A					:		Union[str, Any]						=  '''weight'''
																				elif "bias" in name:
																								__A					:		Tuple						=  '''bias'''
																				else:
																								__A					:		Union[str, Any]						=  None
																				set_recursively(snake_case_       ,snake_case_       ,snake_case_       ,snake_case_       ,snake_case_		)
																continue
								if not is_used:
												unused_weights.append(snake_case_		)

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







def 	__lowercase  (     snake_case_						:						Any       ,snake_case_						:						List[Any]       ,snake_case_						:						str       ,snake_case_						:						List[str]       ,snake_case_						:						Optional[Any]		)     ->Tuple:



				'''simple docstring'''







				__A					:		int						=  full_name.split('''conv_layers.'''		)[-1]
				__A					:		Any						=  name.split('''.'''		)
				__A					:		List[Any]						=  int(items[0]		)
				__A					:		str						=  int(items[1]		)

				if type_id == 0:
								if "bias" in name:
												assert value.shape == feature_extractor.conv_layers[layer_id].conv.bias.data.shape, (
												    F"""{full_name} has size {value.shape}, but"""
												    F""" {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found."""
												)
												__A					:		List[Any]						=  value
												logger.info(F"""Feat extract conv layer {layer_id} was initialized from {full_name}."""		)
								elif "weight" in name:
												assert value.shape == feature_extractor.conv_layers[layer_id].conv.weight.data.shape, (
												    F"""{full_name} has size {value.shape}, but"""
												    F""" {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found."""
												)
												__A					:		str						=  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:
												assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape, (
												    F"""{full_name} has size {value.shape}, but {feature_extractor[layer_id].layer_norm.bias.data.shape} was"""
												    " found."
												)
												__A					:		Union[str, Any]						=  value
												logger.info(F"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""		)
								elif "weight" in name:
												assert value.shape == feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape, (
												    F"""{full_name} has size {value.shape}, but"""
												    F""" {feature_extractor[layer_id].layer_norm.weight.data.shape} was found."""
												)
												__A					:		List[Any]						=  value
												logger.info(F"""Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}."""		)
				else:
								unused_weights.append(snake_case_		)







def 	__lowercase  (     snake_case_						:						Optional[int]		)     ->Dict:



				'''simple docstring'''







				__A					:		List[Any]						=  HubertConfig()
				__A					:		Optional[int]						=  model.config

				__A					:		int						=  fs_config.activation_dropout
				__A					:		str						=  False
				__A					:		Union[str, Any]						=  fs_config.attention_dropout
				__A					:		Union[str, Any]						=  False
				__A					:		str						=  eval(fs_config.extractor_conv_feature_layers		)
				__A					:		Union[str, Any]						=  [x[0] for x in conv_layers]
				__A					:		Union[str, Any]						=  [x[1] for x in conv_layers]
				__A					:		Union[str, Any]						=  [x[2] for x in conv_layers]
				__A					:		Dict						=  '''gelu'''
				__A					:		List[Any]						=  '''layer''' if fs_config.extractor_mode == '''layer_norm''' else '''group'''
				__A					:		int						=  False
				__A					:		int						=  0.0
				__A					:		Dict						=  0.0
				__A					:		str						=  fs_config.activation_fn
				__A					:		List[str]						=  fs_config.dropout
				__A					:		List[Any]						=  fs_config.encoder_embed_dim
				__A					:		Any						=  0.02
				__A					:		List[Any]						=  fs_config.encoder_ffn_embed_dim
				__A					:		Any						=  1e-5
				__A					:		List[Any]						=  0.0
				__A					:		Optional[Any]						=  fs_config.encoder_attention_heads
				__A					:		List[str]						=  fs_config.conv_pos_groups
				__A					:		Optional[Any]						=  fs_config.conv_pos
				__A					:		Dict						=  len(snake_case_		)
				__A					:		int						=  fs_config.encoder_layers

				return config







@torch.no_grad()
def 	__lowercase  (     snake_case_						:						int       ,snake_case_						:						int=None		)     ->Optional[Any]:



				'''simple docstring'''







				__A					:		List[str]						=  distilhubert().model.model

				if config_path is not None:
								__A					:		str						=  HubertConfig.from_pretrained(snake_case_		)
				else:
								__A					:		List[Any]						=  convert_config(snake_case_		)
				__A					:		Any						=  model.eval()

				__A					:		int						=  WavaVecaFeatureExtractor(
				    feature_size=1       ,sampling_rate=16000       ,padding_value=0       ,do_normalize=snake_case_       ,return_attention_mask=snake_case_       ,)
				__A					:		Optional[Any]						=  HubertModel(snake_case_		)

				recursively_load_weights(snake_case_       ,snake_case_		)

				feature_extractor.save_pretrained(snake_case_		)
				hf_model.save_pretrained(snake_case_		)


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("""--config_path""", default=None, type=str, help="""Path to hf config.json of model to convert""")
							a_				=    parser.parse_args()
							convert_hubert_checkpoint(args.pytorch_dump_folder_path, args.config_path)











"""simple docstring"""



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_video_inputs


if is_torch_available():
       import torch

if is_vision_available():
       from PIL import Image

       from transformers import VideoMAEImageProcessor






class 	__snake_case      (	unittest.TestCase  ):





 """simple docstring"""


 def __init__(					self      ,		__lowerCamelCase      ,		__lowerCamelCase=7      ,		__lowerCamelCase=3      ,		__lowerCamelCase=10      ,		__lowerCamelCase=18      ,		__lowerCamelCase=30      ,		__lowerCamelCase=400      ,		__lowerCamelCase=True      ,		__lowerCamelCase=None      ,		__lowerCamelCase=True      ,		__lowerCamelCase=[0.5, 0.5, 0.5]      ,		__lowerCamelCase=[0.5, 0.5, 0.5]      ,		__lowerCamelCase=None      ,		):



     '''simple docstring'''




     __A					:		Optional[Any]						=  size if size is not None else {'''shortest_edge''': 18}
     __A					:		Optional[int]						=  crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}

     __A					:		Any						=  parent
     __A					:		Union[str, Any]						=  batch_size
     __A					:		Optional[int]						=  num_channels
     __A					:		int						=  num_frames
     __A					:		Optional[int]						=  image_size
     __A					:		Union[str, Any]						=  min_resolution
     __A					:		int						=  max_resolution
     __A					:		Tuple						=  do_resize
     __A					:		str						=  size
     __A					:		Any						=  do_normalize
     __A					:		Union[str, Any]						=  image_mean
     __A					:		List[str]						=  image_std
     __A					:		Union[str, Any]						=  crop_size







 def       UpperCamelCase__(					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,
         "crop_size": self.crop_size,
     }






@require_torch
@require_vision
class 	__snake_case      (	SCREAMING_SNAKE_CASE__							,							unittest.TestCase  ):





 """simple docstring"""


 _lowerCamelCase				 =  VideoMAEImageProcessor if is_vision_available() else None

 def       UpperCamelCase__(					self						):



     '''simple docstring'''




     __A					:		List[str]						=  VideoMAEImageProcessingTester(self						)

 @property
 def       UpperCamelCase__(					self						):



     '''simple docstring'''




     return self.image_processor_tester.prepare_image_processor_dict()

 def       UpperCamelCase__(					self						):



     '''simple docstring'''




     __A					:		Dict						=  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      ,		'''do_center_crop'''						)						)
     self.assertTrue(hasattr(__lowerCamelCase      ,		'''size'''						)						)

 def       UpperCamelCase__(					self						):



     '''simple docstring'''




     __A					:		Optional[int]						=  self.image_processing_class.from_dict(self.image_processor_dict						)
     self.assertEqual(image_processor.size      ,		{'''shortest_edge''': 18}						)
     self.assertEqual(image_processor.crop_size      ,		{'''height''': 18, '''width''': 18}						)

     __A					:		Dict						=  self.image_processing_class.from_dict(self.image_processor_dict      ,		size=42      ,		crop_size=84						)
     self.assertEqual(image_processor.size      ,		{'''shortest_edge''': 42}						)
     self.assertEqual(image_processor.crop_size      ,		{'''height''': 84, '''width''': 84}						)

 def       UpperCamelCase__(					self						):



     '''simple docstring'''




     pass

 def       UpperCamelCase__(					self						):



     '''simple docstring'''




     __A					:		List[Any]						=  self.image_processing_class(**self.image_processor_dict						)
     # create random PIL videos
     __A					:		Optional[int]						=  prepare_video_inputs(self.image_processor_tester      ,		equal_resolution=__lowerCamelCase						)
     for video in video_inputs:
         self.assertIsInstance(__lowerCamelCase      ,		__lowerCamelCase						)
         self.assertIsInstance(video[0]      ,		Image.Image						)

     # Test not batched input
     __A					:		Any						=  image_processing(video_inputs[0]      ,		return_tensors='''pt'''						).pixel_values
     self.assertEqual(
         encoded_videos.shape      ,		(
             1,
             self.image_processor_tester.num_frames,
             self.image_processor_tester.num_channels,
             self.image_processor_tester.crop_size['''height'''],
             self.image_processor_tester.crop_size['''width'''],
         )      ,		)

     # Test batched
     __A					:		int						=  image_processing(__lowerCamelCase      ,		return_tensors='''pt'''						).pixel_values
     self.assertEqual(
         encoded_videos.shape      ,		(
             self.image_processor_tester.batch_size,
             self.image_processor_tester.num_frames,
             self.image_processor_tester.num_channels,
             self.image_processor_tester.crop_size['''height'''],
             self.image_processor_tester.crop_size['''width'''],
         )      ,		)

 def       UpperCamelCase__(					self						):



     '''simple docstring'''




     __A					:		List[Any]						=  self.image_processing_class(**self.image_processor_dict						)
     # create random numpy tensors
     __A					:		List[str]						=  prepare_video_inputs(self.image_processor_tester      ,		equal_resolution=__lowerCamelCase      ,		numpify=__lowerCamelCase						)
     for video in video_inputs:
         self.assertIsInstance(__lowerCamelCase      ,		__lowerCamelCase						)
         self.assertIsInstance(video[0]      ,		np.ndarray						)

     # Test not batched input
     __A					:		List[str]						=  image_processing(video_inputs[0]      ,		return_tensors='''pt'''						).pixel_values
     self.assertEqual(
         encoded_videos.shape      ,		(
             1,
             self.image_processor_tester.num_frames,
             self.image_processor_tester.num_channels,
             self.image_processor_tester.crop_size['''height'''],
             self.image_processor_tester.crop_size['''width'''],
         )      ,		)

     # Test batched
     __A					:		int						=  image_processing(__lowerCamelCase      ,		return_tensors='''pt'''						).pixel_values
     self.assertEqual(
         encoded_videos.shape      ,		(
             self.image_processor_tester.batch_size,
             self.image_processor_tester.num_frames,
             self.image_processor_tester.num_channels,
             self.image_processor_tester.crop_size['''height'''],
             self.image_processor_tester.crop_size['''width'''],
         )      ,		)







 def       UpperCamelCase__(					self						):



     '''simple docstring'''




     __A					:		Optional[int]						=  self.image_processing_class(**self.image_processor_dict						)
     # create random PyTorch tensors
     __A					:		int						=  prepare_video_inputs(self.image_processor_tester      ,		equal_resolution=__lowerCamelCase      ,		torchify=__lowerCamelCase						)
     for video in video_inputs:
         self.assertIsInstance(__lowerCamelCase      ,		__lowerCamelCase						)
         self.assertIsInstance(video[0]      ,		torch.Tensor						)

     # Test not batched input
     __A					:		Optional[int]						=  image_processing(video_inputs[0]      ,		return_tensors='''pt'''						).pixel_values
     self.assertEqual(
         encoded_videos.shape      ,		(
             1,
             self.image_processor_tester.num_frames,
             self.image_processor_tester.num_channels,
             self.image_processor_tester.crop_size['''height'''],
             self.image_processor_tester.crop_size['''width'''],
         )      ,		)

     # Test batched
     __A					:		str						=  image_processing(__lowerCamelCase      ,		return_tensors='''pt'''						).pixel_values
     self.assertEqual(
         encoded_videos.shape      ,		(
             self.image_processor_tester.batch_size,
             self.image_processor_tester.num_frames,
             self.image_processor_tester.num_channels,
             self.image_processor_tester.crop_size['''height'''],
             self.image_processor_tester.crop_size['''width'''],
         )      ,		)




