

import os
import warnings
from pathlib import Path
from typing import Optional, Union

from .dynamic_module_utils import custom_object_save
from .tokenization_utils_base import PreTrainedTokenizerBase
from .utils import PushToHubMixin, copy_func, direct_transformers_import, logging


lowerCAmelCase__             =		logging.get_logger(__name__)

# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
lowerCAmelCase__             =		direct_transformers_import(Path(__file__).parent)


lowerCAmelCase__             =		{
    """AutoTokenizer""": """PreTrainedTokenizerBase""",
    """AutoFeatureExtractor""": """FeatureExtractionMixin""",
    """AutoImageProcessor""": """ImageProcessingMixin""",
}







class 						lowercase     (					_lowercase				):


		"""simple docstring"""




		a__       							=	["feature_extractor", "tokenizer"]
		# Names need to be attr_class for attr in attributes
		a__       							=	None
		a__       							=	None
		a__       							=	None


		def __init__(					self						,							*__snake_case						,							**__snake_case):
			# Sanitize args and kwargs
			for key in kwargs:
				if key not in self.attributes:
					raise TypeError(f'''Unexpected keyword argument {key}.''')
			for arg, attribute_name in zip(__snake_case						,							self.attributes):
				if attribute_name in kwargs:
					raise TypeError(f'''Got multiple values for argument {attribute_name}.''')
				else:
					_UpperCamelCase :		Any								=    arg

			if len(__snake_case) != len(self.attributes):
				raise ValueError(
				    f'''This processor requires {len(self.attributes)} arguments: {", ".join(self.attributes)}. Got '''
				    f'''{len(__snake_case)} arguments instead.''')

			# Check each arg is of the proper class (this will also catch a user initializing in the wrong order)
			for attribute_name, arg in kwargs.items():
				_UpperCamelCase :		Any								=    getattr(self						,							f'''{attribute_name}_class''')
				# Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
				_UpperCamelCase :		int								=    AUTO_TO_BASE_CLASS_MAPPING.get(__snake_case						,							__snake_case)
				if isinstance(__snake_case						,							__snake_case):
					_UpperCamelCase :		Union[str, Any]								=    tuple(getattr(__snake_case						,							__snake_case) for n in class_name if n is not None)
				else:
					_UpperCamelCase :		int								=    getattr(__snake_case						,							__snake_case)

				if not isinstance(__snake_case						,							__snake_case):
					raise ValueError(
					    f'''Received a {type(__snake_case).__name__} for argument {attribute_name}, but a {class_name} was expected.''')

				setattr(self						,							__snake_case						,							__snake_case)


		def __repr__(					self):
			_UpperCamelCase :		List[str]								=    [f'''- {name}: {repr(getattr(self						,							__snake_case))}''' for name in self.attributes]
			_UpperCamelCase :		List[str]								=    '\n'.join(__snake_case)
			return f'''{self.__class__.__name__}:\n{attributes_repr}'''


		def 		A__       (					self						,							__snake_case						,							__snake_case = False						,							**__snake_case):
			os.makedirs(__snake_case						,							exist_ok=__snake_case)

			if push_to_hub:
				_UpperCamelCase :		List[Any]								=    kwargs.pop('commit_message'						,							__snake_case)
				_UpperCamelCase :		Tuple								=    kwargs.pop('repo_id'						,							save_directory.split(os.path.sep)[-1])
				_UpperCamelCase :		int								=    self._create_repo(__snake_case						,							**__snake_case)
				_UpperCamelCase :		Tuple								=    self._get_files_timestamps(__snake_case)
			# If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be
			# loaded from the Hub.
			if self._auto_class is not None:
				_UpperCamelCase :		List[str]								=    [getattr(self						,							__snake_case) for attribute_name in self.attributes]
				_UpperCamelCase :		Any								=    [(a.init_kwargs if isinstance(__snake_case						,							__snake_case) else a) for a in attrs]
				custom_object_save(self						,							__snake_case						,							config=__snake_case)

			for attribute_name in self.attributes:
				_UpperCamelCase :		List[str]								=    getattr(self						,							__snake_case)
				# Include the processor class in the attribute config so this processor can then be reloaded with the
				# `AutoProcessor` API.
				if hasattr(__snake_case						,							'_set_processor_class'):
					attribute._set_processor_class(self.__class__.__name__)
				attribute.save_pretrained(__snake_case)

			if self._auto_class is not None:
				# We added an attribute to the init_kwargs of the tokenizers, which needs to be cleaned up.
				for attribute_name in self.attributes:
					_UpperCamelCase :		Tuple								=    getattr(self						,							__snake_case)
					if isinstance(__snake_case						,							__snake_case):
						del attribute.init_kwargs["auto_map"]

			if push_to_hub:
				self._upload_modified_files(
				    __snake_case						,							__snake_case						,							__snake_case						,							commit_message=__snake_case						,							token=kwargs.get('use_auth_token')						,							)


		@classmethod
		def 		A__       (					cls						,							__snake_case						,							__snake_case = None						,							__snake_case = False						,							__snake_case = False						,							__snake_case = None						,							__snake_case = "main"						,							**__snake_case						,							):
			_UpperCamelCase :		Union[str, Any]								=    cache_dir
			_UpperCamelCase :		List[str]								=    force_download
			_UpperCamelCase :		Dict								=    local_files_only
			_UpperCamelCase :		List[Any]								=    revision

			_UpperCamelCase :		Union[str, Any]								=    kwargs.pop('use_auth_token'						,							__snake_case)
			if use_auth_token is not None:
				warnings.warn(
				    'The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers.'						,							__snake_case)
				if token is not None:
					raise ValueError(
					    '`token` and `use_auth_token` are both specified. Please set only the argument `token`.')
				_UpperCamelCase :		Dict								=    use_auth_token

			if token is not None:
				# change to `token` in a follow-up PR
				_UpperCamelCase :		List[str]								=    token

			_UpperCamelCase :		Union[str, Any]								=    cls._get_arguments_from_pretrained(__snake_case						,							**__snake_case)
			return cls(*__snake_case)


		@classmethod
		def 		A__       (					cls						,							__snake_case="AutoProcessor"):
			if not isinstance(__snake_case						,							__snake_case):
				_UpperCamelCase :		Optional[int]								=    auto_class.__name__

			import transformers.models.auto as auto_module

			if not hasattr(__snake_case						,							__snake_case):
				raise ValueError(f'''{auto_class} is not a valid auto class.''')

			_UpperCamelCase :		Optional[int]								=    auto_class


		@classmethod
		def 		A__       (					cls						,							__snake_case						,							**__snake_case):
			_UpperCamelCase :		Any								=    []
			for attribute_name in cls.attributes:
				_UpperCamelCase :		Optional[int]								=    getattr(cls						,							f'''{attribute_name}_class''')
				if isinstance(__snake_case						,							__snake_case):
					_UpperCamelCase :		int								=    tuple(getattr(__snake_case						,							__snake_case) if n is not None else None for n in class_name)
					_UpperCamelCase :		Dict								=    kwargs.get('use_fast'						,							__snake_case)
					if use_fast and classes[1] is not None:
						_UpperCamelCase :		Union[str, Any]								=    classes[1]
					else:
						_UpperCamelCase :		Optional[Any]								=    classes[0]
				else:
					_UpperCamelCase :		Any								=    getattr(__snake_case						,							__snake_case)

				args.append(attribute_class.from_pretrained(__snake_case						,							**__snake_case))
			return args






		@property
		def 		A__       (					self):
			_UpperCamelCase :		str								=    getattr(self						,							self.attributes[0])
			return getattr(__snake_case						,							'model_input_names'						,							__snake_case)

lowerCAmelCase__             =		copy_func(ProcessorMixin.push_to_hub)
if ProcessorMixin.push_to_hub.__doc__ is not None:
							lowerCAmelCase__             =		ProcessorMixin.push_to_hub.__doc__.format(
							    object="""processor""", object_class="""AutoProcessor""", object_files="""processor files"""
							)







import argparse

import torch
from clip import load

from transformers import CLIPConfig, CLIPModel






def        lowerCamelCase_  (				UpperCAmelCase_				:  Optional[int]				,							UpperCAmelCase_				:  Tuple						)    ->					str:




	'''simple docstring'''




	_UpperCamelCase						,		_UpperCamelCase						,		_UpperCamelCase :		Tuple								=    pt_attn_layer.in_proj_weight.chunk(3				,							dim=0						)
	_UpperCamelCase						,		_UpperCamelCase						,		_UpperCamelCase :		str								=    pt_attn_layer.in_proj_bias.chunk(3				,							dim=0						)

	_UpperCamelCase :		Dict								=    pt_attn_layer.out_proj.weight
	_UpperCamelCase :		Tuple								=    pt_attn_layer.out_proj.bias

	_UpperCamelCase :		Tuple								=    q_proj
	_UpperCamelCase :		Dict								=    q_proj_bias

	_UpperCamelCase :		Dict								=    k_proj
	_UpperCamelCase :		str								=    k_proj_bias

	_UpperCamelCase :		Dict								=    v_proj
	_UpperCamelCase :		Optional[int]								=    v_proj_bias

	_UpperCamelCase :		Any								=    out_proj_weights
	_UpperCamelCase :		Optional[int]								=    out_proj_bias






def        lowerCamelCase_  (				UpperCAmelCase_				:  Optional[int]				,							UpperCAmelCase_				:  Union[str, Any]						)    ->					List[Any]:




	'''simple docstring'''




	copy_linear(hf_mlp.fca				,							pt_mlp.c_fc						)
	copy_linear(hf_mlp.fca				,							pt_mlp.c_proj						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  List[str]				,							UpperCAmelCase_				:  List[str]						)    ->					List[Any]:




	'''simple docstring'''




	_UpperCamelCase :		int								=    pt_linear.weight
	_UpperCamelCase :		int								=    pt_linear.bias






def        lowerCamelCase_  (				UpperCAmelCase_				:  str				,							UpperCAmelCase_				:  Union[str, Any]						)    ->					Dict:




	'''simple docstring'''




	copy_linear(hf_layer.layer_norma				,							pt_layer.ln_a						)
	copy_linear(hf_layer.layer_norma				,							pt_layer.ln_a						)

	# copy MLP
	copy_mlp(hf_layer.mlp				,							pt_layer.mlp						)

	# copy attn
	copy_attn_layer(hf_layer.self_attn				,							pt_layer.attn						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  Optional[Any]				,							UpperCAmelCase_				:  List[str]						)    ->					Dict:




	'''simple docstring'''




	for hf_layer, pt_layer in zip(UpperCAmelCase_				,							UpperCAmelCase_						):
		copy_layer(UpperCAmelCase_				,							UpperCAmelCase_						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  Any				,							UpperCAmelCase_				:  Optional[Any]						)    ->					int:




	'''simple docstring'''




	_UpperCamelCase :		Optional[int]								=    pt_model.token_embedding.weight
	_UpperCamelCase :		Tuple								=    pt_model.positional_embedding

	# copy layer norm
	copy_linear(hf_encoder.final_layer_norm				,							pt_model.ln_final						)

	# copy hidden layers
	copy_layers(hf_encoder.encoder.layers				,							pt_model.transformer.resblocks						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  int				,							UpperCAmelCase_				:  Tuple						)    ->					Union[str, Any]:




	'''simple docstring'''




	_UpperCamelCase :		Any								=    pt_model.text_projection.data.T

	# copy text encoder
	copy_encoder(hf_model.text_model				,							UpperCAmelCase_						)






def        lowerCamelCase_  (				UpperCAmelCase_				:  str				,							UpperCAmelCase_				:  Dict						)    ->					Optional[int]:




	'''simple docstring'''




	_UpperCamelCase :		str								=    pt_model.visual.proj.data.T

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

	# copy embeds
	_UpperCamelCase :		Tuple								=    pt_model.visual.conva.weight.data
	_UpperCamelCase :		Dict								=    pt_model.visual.class_embedding
	_UpperCamelCase :		str								=    pt_model.visual.positional_embedding.data

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






@torch.no_grad()
def        lowerCamelCase_  (				UpperCAmelCase_				:  Dict				,							UpperCAmelCase_				:  str				,							UpperCAmelCase_				:  List[str]=None						)    ->					Union[str, Any]:




	'''simple docstring'''




	if config_path is not None:
		_UpperCamelCase :		Dict								=    CLIPConfig.from_pretrained(UpperCAmelCase_						)
	else:
		_UpperCamelCase :		Tuple								=    CLIPConfig(projection_dim=5_1_2				,							text_config={}				,							vision_config={}						)

	_UpperCamelCase :		Tuple								=    CLIPModel(UpperCAmelCase_						).eval()

	_UpperCamelCase						,		_UpperCamelCase :		Optional[Any]								=    load(UpperCAmelCase_				,							device='cpu'				,							jit=UpperCAmelCase_						)
	_UpperCamelCase :		Optional[Any]								=    pt_model.eval()

	copy_text_model_and_projection(UpperCAmelCase_				,							UpperCAmelCase_						)
	copy_vison_model_and_projection(UpperCAmelCase_				,							UpperCAmelCase_						)
	_UpperCamelCase :		Optional[Any]								=    pt_model.logit_scale

	_UpperCamelCase :		Optional[int]								=    torch.arange(0				,							7_7						).unsqueeze(0						)
	_UpperCamelCase :		Dict								=    torch.randn(1				,							3				,							2_2_4				,							2_2_4						)

	_UpperCamelCase :		Optional[Any]								=    hf_model(input_ids=UpperCAmelCase_				,							pixel_values=UpperCAmelCase_				,							return_dict=UpperCAmelCase_						)
	_UpperCamelCase :		Optional[Any]								=    hf_outputs.logits_per_image
	_UpperCamelCase :		Optional[int]								=    hf_outputs.logits_per_text
	_UpperCamelCase						,		_UpperCamelCase :		int								=    pt_model(UpperCAmelCase_				,							UpperCAmelCase_						)

	assert torch.allclose(UpperCAmelCase_				,							UpperCAmelCase_				,							atol=1e-3						)
	assert torch.allclose(UpperCAmelCase_				,							UpperCAmelCase_				,							atol=1e-3						)

	hf_model.save_pretrained(UpperCAmelCase_						)


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

							convert_clip_checkpoint(args.checkpoint_path, args.pytorch_dump_folder_path, args.config_path)




