



"""simple docstring"""






import argparse
import json
import os

import fairseq
import torch
from fairseq.data import Dictionary

from transformers import (
    WavaVecaConfig,
    WavaVecaCTCTokenizer,
    WavaVecaFeatureExtractor,
    WavaVecaForCTC,
    WavaVecaForPreTraining,
    WavaVecaProcessor,
    logging,
)
from transformers.models.wavaveca.modeling_wavaveca import WavaVecaForSequenceClassification


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

lowerCAmelCase__      =    {
    '''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''',
    '''adapter_layer''': '''encoder.layers.*.adapter_layer''',
    '''w2v_model.layer_norm''': '''feature_projection.layer_norm''',
    '''quantizer.weight_proj''': '''quantizer.weight_proj''',
    '''quantizer.vars''': '''quantizer.codevectors''',
    '''project_q''': '''project_q''',
    '''final_proj''': '''project_hid''',
    '''w2v_encoder.proj''': '''lm_head''',
    '''mask_emb''': '''masked_spec_embed''',
    '''pooling_layer.linear''': '''projector''',
    '''pooling_layer.projection''': '''classifier''',
}
lowerCAmelCase__      =    [
    '''lm_head''',
    '''quantizer.weight_proj''',
    '''quantizer.codevectors''',
    '''project_q''',
    '''project_hid''',
    '''projector''',
    '''classifier''',
]






def 	snake_case_  (				A_ :		Tuple						):


							'''simple docstring'''





							_lowerCamelCase						:    str         =			{}
							with open(lowerCAmelCase__,    '''r'''						) as file:
														for line_number, line in enumerate(lowerCAmelCase__						):
																					_lowerCamelCase						:    Dict         =			line.strip()
																					if line:
																												_lowerCamelCase						:    List[str]         =			line.split()
																												_lowerCamelCase						:    Optional[int]         =			line_number
																												_lowerCamelCase						:    Optional[int]         =			words[0]
																												_lowerCamelCase						:    List[Any]         =			value
							return result






def 	snake_case_  (				A_ :		Dict,    A_ :		List[Any],    A_ :		Optional[int],    A_ :		Dict,    A_ :		List[str]						):


							'''simple docstring'''





							for attribute in key.split('''.'''						):
														_lowerCamelCase						:    Union[str, Any]         =			getattr(lowerCAmelCase__,    lowerCAmelCase__						)

							_lowerCamelCase						:    str         =			None
							for param_key in PARAM_MAPPING.keys():
														if full_name.endswith(lowerCAmelCase__						):
																					_lowerCamelCase						:    int         =			PARAM_MAPPING[full_name.split('''.'''						)[-1]]
																					_lowerCamelCase						:    Dict         =			'''param'''

							if weight_type is not None and weight_type != "param":
														_lowerCamelCase						:    Optional[Any]         =			getattr(lowerCAmelCase__,    lowerCAmelCase__						).shape
							elif weight_type is not None and weight_type == "param":
														_lowerCamelCase						:    List[str]         =			hf_pointer
														for attribute in hf_param_name.split('''.'''						):
																					_lowerCamelCase						:    List[Any]         =			getattr(lowerCAmelCase__,    lowerCAmelCase__						)
														_lowerCamelCase						:    int         =			shape_pointer.shape

														# let's reduce dimension
														_lowerCamelCase						:    Union[str, Any]         =			value[0]
							else:
														_lowerCamelCase						:    str         =			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						:    Union[str, Any]         =			value
							elif weight_type == "weight_g":
														_lowerCamelCase						:    Tuple         =			value
							elif weight_type == "weight_v":
														_lowerCamelCase						:    str         =			value
							elif weight_type == "bias":
														_lowerCamelCase						:    int         =			value
							elif weight_type == "param":
														for attribute in hf_param_name.split('''.'''						):
																					_lowerCamelCase						:    int         =			getattr(lowerCAmelCase__,    lowerCAmelCase__						)
														_lowerCamelCase						:    Tuple         =			value
							else:
														_lowerCamelCase						:    List[str]         =			value

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






def 	snake_case_  (				A_ :		Optional[int],    A_ :		Any,    A_ :		int,    A_ :		Any,    A_ :		List[Any]						):


							'''simple docstring'''





							_lowerCamelCase						:    List[Any]         =			None
							for param_key in PARAM_MAPPING.keys():
														if full_name.endswith(lowerCAmelCase__						):
																					_lowerCamelCase						:    str         =			PARAM_MAPPING[full_name.split('''.'''						)[-1]]
																					_lowerCamelCase						:    Optional[Any]         =			'''param'''

							if weight_type is not None and weight_type != "param":
														_lowerCamelCase						:    Optional[int]         =			'''.'''.join([key, weight_type]						)
							elif weight_type is not None and weight_type == "param":
														_lowerCamelCase						:    int         =			'''.'''.join([key, hf_param_name]						)
							else:
														_lowerCamelCase						:    List[str]         =			key

							_lowerCamelCase						:    List[str]         =			value if '''lm_head''' in full_key else value[0]


lowerCAmelCase__      =    {
    '''W_a''': '''linear_1.weight''',
    '''W_b''': '''linear_2.weight''',
    '''b_a''': '''linear_1.bias''',
    '''b_b''': '''linear_2.bias''',
    '''ln_W''': '''norm.weight''',
    '''ln_b''': '''norm.bias''',
}






def 	snake_case_  (				A_ :		str,    A_ :		Tuple,    A_ :		Optional[int]=None,    A_ :		str=None						):


							'''simple docstring'''





							_lowerCamelCase						:    int         =			False
							for key, mapped_key in MAPPING.items():
														_lowerCamelCase						:    Union[str, Any]         =			'''wav2vec2.''' + 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						:    Optional[Any]         =			True
																					if "*" in mapped_key:
																												_lowerCamelCase						:    Tuple         =			name.split(lowerCAmelCase__						)[0].split('''.'''						)[-2]
																												_lowerCamelCase						:    Dict         =			mapped_key.replace('''*''',    lowerCAmelCase__						)
																					if "weight_g" in name:
																												_lowerCamelCase						:    Optional[int]         =			'''weight_g'''
																					elif "weight_v" in name:
																												_lowerCamelCase						:    Optional[int]         =			'''weight_v'''
																					elif "bias" in name:
																												_lowerCamelCase						:    Dict         =			'''bias'''
																					elif "weight" in name:
																												# TODO: don't match quantizer.weight_proj
																												_lowerCamelCase						:    List[str]         =			'''weight'''
																					else:
																												_lowerCamelCase						:    Union[str, Any]         =			None
																					if hf_dict is not None:
																												rename_dict(lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__						)
																					else:
																												set_recursively(lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__						)
																					return is_used
							return is_used






def 	snake_case_  (				A_ :		Tuple,    A_ :		Tuple,    A_ :		Optional[int]						):


							'''simple docstring'''





							_lowerCamelCase						:    List[str]         =			[]
							_lowerCamelCase						:    Any         =			fairseq_model.state_dict()

							_lowerCamelCase						:    int         =			hf_model.wavaveca.feature_extractor

							for name, value in fairseq_dict.items():
														_lowerCamelCase						:    Optional[int]         =			False
														if "conv_layers" in name:
																					load_conv_layer(
																					    lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__,    hf_model.config.feat_extract_norm == '''group''',    )
																					_lowerCamelCase						:    Optional[Any]         =			True
														else:
																					_lowerCamelCase						:    Optional[int]         =			load_wavaveca_layer(lowerCAmelCase__,    lowerCAmelCase__,    lowerCAmelCase__						)
														if not is_used:
																					unused_weights.append(lowerCAmelCase__						)

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






def 	snake_case_  (				A_ :		str,    A_ :		Union[str, Any],    A_ :		List[str],    A_ :		Tuple,    A_ :		str						):


							'''simple docstring'''





							_lowerCamelCase						:    Tuple         =			full_name.split('''conv_layers.'''						)[-1]
							_lowerCamelCase						:    Optional[int]         =			name.split('''.'''						)
							_lowerCamelCase						:    int         =			int(items[0]						)
							_lowerCamelCase						:    Dict         =			int(items[1]						)

							if type_id == 0:
														if "bias" in name:
																					if value.shape != feature_extractor.conv_layers[layer_id].conv.bias.data.shape:
																												raise ValueError(
																												    F'''{full_name} has size {value.shape}, but'''
																												    F''' {feature_extractor.conv_layers[layer_id].conv.bias.data.shape} was found.'''						)
																					_lowerCamelCase						:    Union[str, Any]         =			value
																					logger.info(F'''Feat extract conv layer {layer_id} was initialized from {full_name}.'''						)
														elif "weight" in name:
																					if value.shape != feature_extractor.conv_layers[layer_id].conv.weight.data.shape:
																												raise ValueError(
																												    F'''{full_name} has size {value.shape}, but'''
																												    F''' {feature_extractor.conv_layers[layer_id].conv.weight.data.shape} was found.'''						)
																					_lowerCamelCase						:    Optional[int]         =			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:
																					if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape:
																												raise ValueError(
																												    F'''{full_name} has size {value.shape}, but'''
																												    F''' {feature_extractor.conv_layers[layer_id].layer_norm.bias.data.shape} was found.'''						)
																					_lowerCamelCase						:    int         =			value
																					logger.info(F'''Feat extract layer norm weight of layer {layer_id} was initialized from {full_name}.'''						)
														elif "weight" in name:
																					if value.shape != feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape:
																												raise ValueError(
																												    F'''{full_name} has size {value.shape}, but'''
																												    F''' {feature_extractor.conv_layers[layer_id].layer_norm.weight.data.shape} was found.'''						)
																					_lowerCamelCase						:    Optional[int]         =			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 	snake_case_  (				A_ :		Union[str, Any],    A_ :		List[Any],    A_ :		Tuple=None,    A_ :		Union[str, Any]=None,    A_ :		int=True,    A_ :		Tuple=False						):


							'''simple docstring'''





							if config_path is not None:
														_lowerCamelCase						:    Dict         =			WavaVecaConfig.from_pretrained(lowerCAmelCase__						)
							else:
														_lowerCamelCase						:    Dict         =			WavaVecaConfig()

							if is_seq_class:
														_lowerCamelCase						:    str         =			read_txt_into_dict(lowerCAmelCase__						)
														_lowerCamelCase						:    str         =			idalabel
														_lowerCamelCase						:    str         =			WavaVecaForSequenceClassification(lowerCAmelCase__						)
														_lowerCamelCase						:    str         =			WavaVecaFeatureExtractor(
														    feature_size=1,    sampling_rate=1_60_00,    padding_value=0,    do_normalize=lowerCAmelCase__,    return_attention_mask=lowerCAmelCase__,    )
														feature_extractor.save_pretrained(lowerCAmelCase__						)

							elif is_finetuned:
														if dict_path:
																					_lowerCamelCase						:    Tuple         =			Dictionary.load(lowerCAmelCase__						)

																					# important change bos & pad token id since CTC symbol is <pad> and
																					# not <s> as in fairseq
																					_lowerCamelCase						:    Optional[Any]         =			target_dict.pad_index
																					_lowerCamelCase						:    Tuple         =			target_dict.bos_index
																					_lowerCamelCase						:    int         =			target_dict.eos_index
																					_lowerCamelCase						:    Any         =			len(target_dict.symbols						)
																					_lowerCamelCase						:    Any         =			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__						)
																					_lowerCamelCase						:    Dict         =			target_dict.indices

																					# fairseq has the <pad> and <s> switched
																					_lowerCamelCase						:    str         =			0
																					_lowerCamelCase						:    Any         =			1
																					with open(lowerCAmelCase__,    '''w''',    encoding='''utf-8'''						) as vocab_handle:
																												json.dump(lowerCAmelCase__,    lowerCAmelCase__						)
																					_lowerCamelCase						:    Tuple         =			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__,    )
																					_lowerCamelCase						:    Any         =			True if config.feat_extract_norm == '''layer''' else False
																					_lowerCamelCase						:    Union[str, Any]         =			WavaVecaFeatureExtractor(
																					    feature_size=1,    sampling_rate=1_60_00,    padding_value=0,    do_normalize=lowerCAmelCase__,    return_attention_mask=lowerCAmelCase__,    )
																					_lowerCamelCase						:    Dict         =			WavaVecaProcessor(feature_extractor=lowerCAmelCase__,    tokenizer=lowerCAmelCase__						)
																					processor.save_pretrained(lowerCAmelCase__						)

														_lowerCamelCase						:    Tuple         =			WavaVecaForCTC(lowerCAmelCase__						)
							else:
														_lowerCamelCase						:    List[str]         =			WavaVecaForPreTraining(lowerCAmelCase__						)

							if is_finetuned or is_seq_class:
														_lowerCamelCase						:    int         =			fairseq.checkpoint_utils.load_model_ensemble_and_task(
														    [checkpoint_path],    arg_overrides={'''data''': '''/'''.join(dict_path.split('''/'''						)[:-1]						)}						)
							else:
														_lowerCamelCase						:    List[str]         =			argparse.Namespace(task='''audio_pretraining'''						)
														_lowerCamelCase						:    int         =			fairseq.tasks.setup_task(lowerCAmelCase__						)

														_lowerCamelCase						:    List[Any]         =			fairseq.checkpoint_utils.load_model_ensemble_and_task([checkpoint_path],    task=lowerCAmelCase__						)

							_lowerCamelCase						:    int         =			model[0].eval()

							recursively_load_weights(lowerCAmelCase__,    lowerCAmelCase__,    not is_finetuned						)

							hf_wavavec.save_pretrained(lowerCAmelCase__						)


if __name__ == "__main__":
			lowerCAmelCase__      =    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'''
			)
			parser.add_argument(
			    '''--is_seq_class''',
			    action='''store_true''',
			    help='''Whether the model to convert is a fine-tuned sequence classification model or not''',
			)
			lowerCAmelCase__      =    parser.parse_args()

			lowerCAmelCase__      =    not args.not_finetuned and not args.is_seq_class
			convert_wavaveca_checkpoint(
			    args.checkpoint_path,
			    args.pytorch_dump_folder_path,
			    args.config_path,
			    args.dict_path,
			    is_finetuned,
			    args.is_seq_class,
			)







"""simple docstring"""






from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
   from PIL import Image

   from ..image_utils import load_image

if is_torch_available():
   import torch

   from transformers.modeling_outputs import BaseModelOutput

   from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

lowerCAmelCase__      =    logging.get_logger(__name__)






@add_end_docstrings(_lowercase)
class 	__snake_case	(     _lowercase):




  def __init__(			self					:						Any						, **__lowerCAmelCase					:						Union[str, Any]		):






         """simple docstring"""






         super().__init__(**__lowerCAmelCase		)

         if self.framework == "tf":
                raise ValueError(f'''The {self.__class__} is only available in PyTorch.'''		)

         requires_backends(self						, '''vision'''		)
         self.check_model_type(__lowerCAmelCase		)




  def __call__(			self					:						Dict						, __lowerCAmelCase					:						Union[str, "Image.Image", List[Dict[str, Any]]]						, __lowerCAmelCase					:						Union[str, List[str]] = None						, **__lowerCAmelCase					:						int						, ):






         """simple docstring"""






         if "text_queries" in kwargs:
                _lowerCamelCase						:    List[Any]         =			kwargs.pop('''text_queries'''		)

         if isinstance(__lowerCAmelCase						, (str, Image.Image)		):
                _lowerCamelCase						:    Optional[int]         =			{'''image''': image, '''candidate_labels''': candidate_labels}
         else:
                _lowerCamelCase						:    List[Any]         =			image
         _lowerCamelCase						:    List[str]         =			super().__call__(__lowerCAmelCase						, **__lowerCAmelCase		)
         return results




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]						, **__lowerCAmelCase					:						int		):






         """simple docstring"""






         _lowerCamelCase						:    int         =			{}
         if "threshold" in kwargs:
                _lowerCamelCase						:    Optional[Any]         =			kwargs['''threshold''']
         if "top_k" in kwargs:
                _lowerCamelCase						:    int         =			kwargs['''top_k''']
         return {}, {}, postprocess_params




  def 						SCREAMING_SNAKE_CASE    (			self					:						List[Any]						, __lowerCAmelCase					:						Union[str, Any]		):






         """simple docstring"""






         _lowerCamelCase						:    int         =			load_image(inputs['''image''']		)
         _lowerCamelCase						:    Optional[Any]         =			inputs['''candidate_labels''']
         if isinstance(__lowerCAmelCase						, __lowerCAmelCase		):
                _lowerCamelCase						:    int         =			candidate_labels.split(''','''		)

         _lowerCamelCase						:    Tuple         =			torch.tensor([[image.height, image.width]]						, dtype=torch.intaa		)
         for i, candidate_label in enumerate(__lowerCAmelCase		):
                _lowerCamelCase						:    Any         =			self.tokenizer(__lowerCAmelCase						, return_tensors=self.framework		)
                _lowerCamelCase						:    Optional[Any]         =			self.image_processor(__lowerCAmelCase						, return_tensors=self.framework		)
                yield {
                    "is_last": i == len(__lowerCAmelCase		) - 1,
                    "target_size": target_size,
                    "candidate_label": candidate_label,
                    **text_inputs,
                    **image_features,
                }




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any						, __lowerCAmelCase					:						List[Any]		):






         """simple docstring"""






         _lowerCamelCase						:    Optional[Any]         =			model_inputs.pop('''target_size'''		)
         _lowerCamelCase						:    List[Any]         =			model_inputs.pop('''candidate_label'''		)
         _lowerCamelCase						:    Dict         =			model_inputs.pop('''is_last'''		)

         _lowerCamelCase						:    str         =			self.model(**__lowerCAmelCase		)

         _lowerCamelCase						:    Optional[Any]         =			{'''target_size''': target_size, '''candidate_label''': candidate_label, '''is_last''': is_last, **outputs}
         return model_outputs




  def 						SCREAMING_SNAKE_CASE    (			self					:						Optional[Any]						, __lowerCAmelCase					:						Any						, __lowerCAmelCase					:						Union[str, Any]=0.1						, __lowerCAmelCase					:						Optional[Any]=None		):






         """simple docstring"""






         _lowerCamelCase						:    str         =			[]
         for model_output in model_outputs:
                _lowerCamelCase						:    Any         =			model_output['''candidate_label''']
                _lowerCamelCase						:    Union[str, Any]         =			BaseModelOutput(__lowerCAmelCase		)
                _lowerCamelCase						:    Tuple         =			self.image_processor.post_process_object_detection(
                    outputs=__lowerCAmelCase						, threshold=__lowerCAmelCase						, target_sizes=model_output['''target_size''']		)[0]

                for index in outputs["scores"].nonzero():
                       _lowerCamelCase						:    Tuple         =			outputs['''scores'''][index].item()
                       _lowerCamelCase						:    Optional[Any]         =			self._get_bounding_box(outputs['''boxes'''][index][0]		)

                       _lowerCamelCase						:    Optional[Any]         =			{'''score''': score, '''label''': label, '''box''': box}
                       results.append(__lowerCAmelCase		)

         _lowerCamelCase						:    int         =			sorted(__lowerCAmelCase						, key=lambda __lowerCAmelCase		: x["score"]						, reverse=__lowerCAmelCase		)
         if top_k:
                _lowerCamelCase						:    Dict         =			results[:top_k]

         return results




  def 						SCREAMING_SNAKE_CASE    (			self					:						Any						, __lowerCAmelCase					:						"torch.Tensor"		):






         """simple docstring"""






         if self.framework != "pt":
                raise ValueError('''The ZeroShotObjectDetectionPipeline is only available in PyTorch.'''		)
         _lowerCamelCase      ,   _lowerCamelCase      ,   _lowerCamelCase      ,   _lowerCamelCase						:    Dict         =			box.int().tolist()
         _lowerCamelCase						:    Union[str, Any]         =			{
             '''xmin''': xmin,
             '''ymin''': ymin,
             '''xmax''': xmax,
             '''ymax''': ymax,
         }
         return bbox


