

'''simple docstring'''

import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    HubertConfig,
    HubertForCTC,
    HubertModel,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaProcessor,
    logging,
)


logging.set_verbosity_info()
SCREAMING_SNAKE_CASE							:				Optional[int]     	=			logging.get_logger(__name__)

SCREAMING_SNAKE_CASE							:				List[str]     	=			{
    "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",
    "w2v_model.layer_norm": "feature_projection.layer_norm",
    "w2v_encoder.proj": "lm_head",
    "mask_emb": "masked_spec_embed",
}




def 							_UpperCamelCase				(   lowerCAmelCase__:					List[Any]		,lowerCAmelCase__:					Union[str, Any]		,lowerCAmelCase__:					List[str]		,lowerCAmelCase__:					Tuple		,lowerCAmelCase__:					Dict   )      -> str:
							for attribute in key.split('.'   ):
														SCREAMING_SNAKE_CASE_				=				getattr(lowerCAmelCase__		,lowerCAmelCase__   )

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

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




def 							_UpperCamelCase				(   lowerCAmelCase__:					Optional[Any]		,lowerCAmelCase__:					List[str]		,lowerCAmelCase__:					List[str]   )      -> Dict:
							SCREAMING_SNAKE_CASE_				=				[]
							SCREAMING_SNAKE_CASE_				=				fairseq_model.state_dict()

							SCREAMING_SNAKE_CASE_				=				hf_model.hubert.feature_extractor if is_finetuned else hf_model.feature_extractor

							for name, value in fairseq_dict.items():
														SCREAMING_SNAKE_CASE_				=				False
														if "conv_layers" in name:
																					load_conv_layer(
																					    lowerCAmelCase__		,lowerCAmelCase__		,lowerCAmelCase__		,lowerCAmelCase__		,hf_model.config.feat_extract_norm == 'group'		,)
																					SCREAMING_SNAKE_CASE_				=				True
														else:
																					for key, mapped_key in MAPPING.items():
																												SCREAMING_SNAKE_CASE_				=				'hubert.' + mapped_key if (is_finetuned and mapped_key != 'lm_head') else mapped_key

																												if key in name or (key.split('w2v_model.'   )[-1] == name.split('.'   )[0] and not is_finetuned):
																																			SCREAMING_SNAKE_CASE_				=				True
																																			if "*" in mapped_key:
																																										SCREAMING_SNAKE_CASE_				=				name.split(lowerCAmelCase__   )[0].split('.'   )[-2]
																																										SCREAMING_SNAKE_CASE_				=				mapped_key.replace('*'		,lowerCAmelCase__   )
																																			if "weight_g" in name:
																																										SCREAMING_SNAKE_CASE_				=				'weight_g'
																																			elif "weight_v" in name:
																																										SCREAMING_SNAKE_CASE_				=				'weight_v'
																																			elif "weight" in name:
																																										SCREAMING_SNAKE_CASE_				=				'weight'
																																			elif "bias" in name:
																																										SCREAMING_SNAKE_CASE_				=				'bias'
																																			else:
																																										SCREAMING_SNAKE_CASE_				=				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 							_UpperCamelCase				(   lowerCAmelCase__:					Tuple		,lowerCAmelCase__:					List[Any]		,lowerCAmelCase__:					Any		,lowerCAmelCase__:					Dict		,lowerCAmelCase__:					List[str]   )      -> Any:
							SCREAMING_SNAKE_CASE_				=				full_name.split('conv_layers.'   )[-1]
							SCREAMING_SNAKE_CASE_				=				name.split('.'   )
							SCREAMING_SNAKE_CASE_				=				int(items[0]   )
							SCREAMING_SNAKE_CASE_				=				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."""
																					)
																					SCREAMING_SNAKE_CASE_				=				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."""
																					)
																					SCREAMING_SNAKE_CASE_				=				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."
																					)
																					SCREAMING_SNAKE_CASE_				=				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."""
																					)
																					SCREAMING_SNAKE_CASE_				=				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 							_UpperCamelCase				(   lowerCAmelCase__:					Optional[int]		,lowerCAmelCase__:					List[str]		,lowerCAmelCase__:					Union[str, Any]=None		,lowerCAmelCase__:					List[str]=None		,lowerCAmelCase__:					Union[str, Any]=True   )      -> Tuple:
							if config_path is not None:
														SCREAMING_SNAKE_CASE_				=				HubertConfig.from_pretrained(lowerCAmelCase__   )
							else:
														SCREAMING_SNAKE_CASE_				=				HubertConfig()

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

																					# important change bos & pad token id since CTC symbol is <pad> and
																					# not <s> as in fairseq
																					SCREAMING_SNAKE_CASE_				=				target_dict.pad_index
																					SCREAMING_SNAKE_CASE_				=				target_dict.bos_index
																					SCREAMING_SNAKE_CASE_				=				target_dict.eos_index
																					SCREAMING_SNAKE_CASE_				=				len(target_dict.symbols   )
																					SCREAMING_SNAKE_CASE_				=				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__   )
																					with open(lowerCAmelCase__		,'w'		,encoding='utf-8'   ) as vocab_handle:
																												json.dump(target_dict.indices		,lowerCAmelCase__   )
																					SCREAMING_SNAKE_CASE_				=				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__		,)
																					SCREAMING_SNAKE_CASE_				=				True if config.feat_extract_norm == 'layer' else False
																					SCREAMING_SNAKE_CASE_				=				WavaVecaFeatureExtractor(
																					    feature_size=1		,sampling_rate=1_6000		,padding_value=0		,do_normalize=lowerCAmelCase__		,return_attention_mask=lowerCAmelCase__		,)
																					SCREAMING_SNAKE_CASE_				=				WavaVecaProcessor(feature_extractor=lowerCAmelCase__		,tokenizer=lowerCAmelCase__   )
																					processor.save_pretrained(lowerCAmelCase__   )

														SCREAMING_SNAKE_CASE_				=				HubertForCTC(lowerCAmelCase__   )
							else:
														SCREAMING_SNAKE_CASE_				=				HubertModel(lowerCAmelCase__   )

							if is_finetuned:
														SCREAMING_SNAKE_CASE_				,    SCREAMING_SNAKE_CASE_				,    SCREAMING_SNAKE_CASE_				=				fairseq.checkpoint_utils.load_model_ensemble_and_task(
														    [checkpoint_path]		,arg_overrides={'data': '/'.join(dict_path.split('/'   )[:-1]   )}   )
							else:
														SCREAMING_SNAKE_CASE_				,    SCREAMING_SNAKE_CASE_				,    SCREAMING_SNAKE_CASE_				=				fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path]   )

							SCREAMING_SNAKE_CASE_				=				model[0].eval()

							recursively_load_weights(lowerCAmelCase__		,lowerCAmelCase__		,lowerCAmelCase__   )

							hf_wavavec.save_pretrained(lowerCAmelCase__   )


if __name__ == "__main__":
							SCREAMING_SNAKE_CASE							:				Tuple     	=			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"
							)
							SCREAMING_SNAKE_CASE							:				List[Any]     	=			parser.parse_args()
							convert_hubert_checkpoint(
							    args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path, args.dict_path, not args.not_finetuned
							)








'''simple docstring'''

import unittest

from transformers import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING, is_vision_available, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_vision_available():
							from PIL import Image
else:






							class    snake_case     :






											"""simple docstring"""



											@staticmethod
											def   a__					(    *_lowercase,     **_lowercase       )     ->							Tuple:
																		pass






@is_pipeline_test
@require_vision
@require_torch
class    snake_case     (      unittest.TestCase		):






				"""simple docstring"""
				_a    	=     MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING



				def   a__					(    self,     _lowercase,     _lowercase,     _lowercase       )     ->							Tuple:
											SCREAMING_SNAKE_CASE_				=				pipeline(
											    'zero-shot-object-detection',     model='hf-internal-testing/tiny-random-owlvit-object-detection'       )

											SCREAMING_SNAKE_CASE_				=				[
											    {
											        'image': './tests/fixtures/tests_samples/COCO/000000039769.png',
											        'candidate_labels': ['cat', 'remote', 'couch'],
											    }
											]
											return object_detector, examples



				def   a__					(    self,     _lowercase,     _lowercase       )     ->							str:
											SCREAMING_SNAKE_CASE_				=				object_detector(examples[0],     threshold=0.0       )

											SCREAMING_SNAKE_CASE_				=				len(_lowercase       )
											self.assertGreater(_lowercase,     0       )
											self.assertEqual(
											    _lowercase,     [
											        {
											            'score': ANY(_lowercase       ),
											            'label': ANY(_lowercase       ),
											            'box': {'xmin': ANY(_lowercase       ), 'ymin': ANY(_lowercase       ), 'xmax': ANY(_lowercase       ), 'ymax': ANY(_lowercase       )},
											        }
											        for i in range(_lowercase       )
											    ],     )



				@require_tf
				@unittest.skip('Zero Shot Object Detection not implemented in TF'       )
				def   a__					(    self       )     ->							Union[str, Any]:
											pass



				@require_torch
				def   a__					(    self       )     ->							str:
											SCREAMING_SNAKE_CASE_				=				pipeline(
											    'zero-shot-object-detection',     model='hf-internal-testing/tiny-random-owlvit-object-detection'       )

											SCREAMING_SNAKE_CASE_				=				object_detector(
											    './tests/fixtures/tests_samples/COCO/000000039769.png',     candidate_labels=['cat', 'remote', 'couch'],     threshold=0.64,     )

											self.assertEqual(
											    nested_simplify(_lowercase,     decimals=4       ),     [
											        {'score': 0.7_235, 'label': 'cat', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
											        {'score': 0.7_218, 'label': 'remote', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
											        {'score': 0.7_184, 'label': 'couch', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
											        {'score': 0.6_748, 'label': 'remote', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
											        {'score': 0.6_656, 'label': 'cat', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
											        {'score': 0.6_614, 'label': 'couch', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
											        {'score': 0.6_456, 'label': 'remote', 'box': {'xmin': 494, 'ymin': 105, 'xmax': 521, 'ymax': 127}},
											        {'score': 0.642, 'label': 'remote', 'box': {'xmin': 67, 'ymin': 274, 'xmax': 93, 'ymax': 297}},
											        {'score': 0.6_419, 'label': 'cat', 'box': {'xmin': 494, 'ymin': 105, 'xmax': 521, 'ymax': 127}},
											    ],     )

											SCREAMING_SNAKE_CASE_				=				object_detector(
											    [
											        {
											            'image': './tests/fixtures/tests_samples/COCO/000000039769.png',
											            'candidate_labels': ['cat', 'remote', 'couch'],
											        }
											    ],     threshold=0.64,     )

											self.assertEqual(
											    nested_simplify(_lowercase,     decimals=4       ),     [
											        [
											            {'score': 0.7_235, 'label': 'cat', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
											            {'score': 0.7_218, 'label': 'remote', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
											            {'score': 0.7_184, 'label': 'couch', 'box': {'xmin': 204, 'ymin': 167, 'xmax': 232, 'ymax': 190}},
											            {'score': 0.6_748, 'label': 'remote', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
											            {'score': 0.6_656, 'label': 'cat', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
											            {'score': 0.6_614, 'label': 'couch', 'box': {'xmin': 571, 'ymin': 83, 'xmax': 598, 'ymax': 103}},
											            {'score': 0.6_456, 'label': 'remote', 'box': {'xmin': 494, 'ymin': 105, 'xmax': 521, 'ymax': 127}},
											            {'score': 0.642, 'label': 'remote', 'box': {'xmin': 67, 'ymin': 274, 'xmax': 93, 'ymax': 297}},
											            {'score': 0.6_419, 'label': 'cat', 'box': {'xmin': 494, 'ymin': 105, 'xmax': 521, 'ymax': 127}},
											        ]
											    ],     )



				@require_torch
				@slow
				def   a__					(    self       )     ->							Optional[int]:
											SCREAMING_SNAKE_CASE_				=				pipeline('zero-shot-object-detection'       )

											SCREAMING_SNAKE_CASE_				=				object_detector(
											    'http://images.cocodataset.org/val2017/000000039769.jpg',     candidate_labels=['cat', 'remote', 'couch'],     )
											self.assertEqual(
											    nested_simplify(_lowercase,     decimals=4       ),     [
											        {'score': 0.2_868, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
											        {'score': 0.277, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
											        {'score': 0.2_537, 'label': 'cat', 'box': {'xmin': 1, 'ymin': 55, 'xmax': 315, 'ymax': 472}},
											        {'score': 0.1_474, 'label': 'remote', 'box': {'xmin': 335, 'ymin': 74, 'xmax': 371, 'ymax': 187}},
											        {'score': 0.1_208, 'label': 'couch', 'box': {'xmin': 4, 'ymin': 0, 'xmax': 642, 'ymax': 476}},
											    ],     )

											SCREAMING_SNAKE_CASE_				=				object_detector(
											    [
											        {
											            'image': 'http://images.cocodataset.org/val2017/000000039769.jpg',
											            'candidate_labels': ['cat', 'remote', 'couch'],
											        },
											        {
											            'image': 'http://images.cocodataset.org/val2017/000000039769.jpg',
											            'candidate_labels': ['cat', 'remote', 'couch'],
											        },
											    ],     )
											self.assertEqual(
											    nested_simplify(_lowercase,     decimals=4       ),     [
											        [
											            {'score': 0.2_868, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
											            {'score': 0.277, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
											            {'score': 0.2_537, 'label': 'cat', 'box': {'xmin': 1, 'ymin': 55, 'xmax': 315, 'ymax': 472}},
											            {'score': 0.1_474, 'label': 'remote', 'box': {'xmin': 335, 'ymin': 74, 'xmax': 371, 'ymax': 187}},
											            {'score': 0.1_208, 'label': 'couch', 'box': {'xmin': 4, 'ymin': 0, 'xmax': 642, 'ymax': 476}},
											        ],
											        [
											            {'score': 0.2_868, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
											            {'score': 0.277, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
											            {'score': 0.2_537, 'label': 'cat', 'box': {'xmin': 1, 'ymin': 55, 'xmax': 315, 'ymax': 472}},
											            {'score': 0.1_474, 'label': 'remote', 'box': {'xmin': 335, 'ymin': 74, 'xmax': 371, 'ymax': 187}},
											            {'score': 0.1_208, 'label': 'couch', 'box': {'xmin': 4, 'ymin': 0, 'xmax': 642, 'ymax': 476}},
											        ],
											    ],     )



				@require_tf
				@unittest.skip('Zero Shot Object Detection not implemented in TF'       )
				def   a__					(    self       )     ->							Optional[Any]:
											pass



				@require_torch
				@slow
				def   a__					(    self       )     ->							Any:
											SCREAMING_SNAKE_CASE_				=				0.2
											SCREAMING_SNAKE_CASE_				=				pipeline('zero-shot-object-detection'       )

											SCREAMING_SNAKE_CASE_				=				object_detector(
											    'http://images.cocodataset.org/val2017/000000039769.jpg',     candidate_labels=['cat', 'remote', 'couch'],     threshold=_lowercase,     )
											self.assertEqual(
											    nested_simplify(_lowercase,     decimals=4       ),     [
											        {'score': 0.2_868, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
											        {'score': 0.277, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
											        {'score': 0.2_537, 'label': 'cat', 'box': {'xmin': 1, 'ymin': 55, 'xmax': 315, 'ymax': 472}},
											    ],     )





				@require_torch
				@slow
				def   a__					(    self       )     ->							int:
											SCREAMING_SNAKE_CASE_				=				2
											SCREAMING_SNAKE_CASE_				=				pipeline('zero-shot-object-detection'       )

											SCREAMING_SNAKE_CASE_				=				object_detector(
											    'http://images.cocodataset.org/val2017/000000039769.jpg',     candidate_labels=['cat', 'remote', 'couch'],     top_k=_lowercase,     )
											self.assertEqual(
											    nested_simplify(_lowercase,     decimals=4       ),     [
											        {'score': 0.2_868, 'label': 'cat', 'box': {'xmin': 324, 'ymin': 20, 'xmax': 640, 'ymax': 373}},
											        {'score': 0.277, 'label': 'remote', 'box': {'xmin': 40, 'ymin': 72, 'xmax': 177, 'ymax': 115}},
											    ],     )





