



"""simple docstring"""
import gc
import os
import tempfile
import unittest

import torch
import torch.nn as nn
import torch.nn.functional as F
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import AutoencoderKL, DDIMScheduler, StableDiffusionPipeline, UNetaDConditionModel
from diffusers.loaders import AttnProcsLayers, LoraLoaderMixin
from diffusers.models.attention_processor import (
    Attention,
    AttnProcessor,
    AttnProcessor2_0,
    LoRAAttnProcessor,
    LoRAAttnProcessor2_0,
    LoRAXFormersAttnProcessor,
    XFormersAttnProcessor,
)
from diffusers.utils import TEXT_ENCODER_ATTN_MODULE, floats_tensor, torch_device
def   A_    (     _lowercase      ):







						'''simple docstring'''


						snake_case_					:Optional[Any]		=     {}
						for name in unet.attn_processors.keys():
												snake_case_					:Optional[int]		=     None if name.endswith("""attn1.processor"""      ) else unet.config.cross_attention_dim
												if name.startswith("""mid_block"""      ):
																		snake_case_					:Tuple		=     unet.config.block_out_channels[-1]
												elif name.startswith("""up_blocks"""      ):
																		snake_case_					:Any		=     int(name[len("""up_blocks."""      )]      )
																		snake_case_					:Optional[Any]		=     list(reversed(unet.config.block_out_channels      )      )[block_id]
												elif name.startswith("""down_blocks"""      ):
																		snake_case_					:Tuple		=     int(name[len("""down_blocks."""      )]      )
																		snake_case_					:Optional[Any]		=     unet.config.block_out_channels[block_id]
												snake_case_					:int		=     (
												    LoRAAttnProcessor2_0 if hasattr(_lowercase,   """scaled_dot_product_attention"""      ) else LoRAAttnProcessor
												)
												snake_case_					:List[Any]		=     lora_attn_processor_class(
												    hidden_size=_lowercase,   cross_attention_dim=_lowercase      )
						snake_case_					:int		=     AttnProcsLayers(_lowercase      )
						return lora_attn_procs, unet_lora_layers
def   A_    (     _lowercase      ):







						'''simple docstring'''


						snake_case_					:str		=     {}
						snake_case_					:int		=     (
						    LoRAAttnProcessor2_0 if hasattr(_lowercase,   """scaled_dot_product_attention"""      ) else LoRAAttnProcessor
						)
						for name, module in text_encoder.named_modules():
												if name.endswith(_lowercase      ):
																		snake_case_					:Optional[Any]		=     lora_attn_processor_class(
																		    hidden_size=module.out_proj.out_features,   cross_attention_dim=_lowercase      )
						return text_lora_attn_procs
def   A_    (     _lowercase      ):







						'''simple docstring'''


						snake_case_					:Optional[int]		=     create_text_encoder_lora_attn_procs(_lowercase      )
						snake_case_					:int		=     AttnProcsLayers(_lowercase      )
						return text_encoder_lora_layers
def   A_    (     _lowercase,   _lowercase=False      ):







						'''simple docstring'''


						for _, attn_proc in text_lora_attn_procs.items():
												# set up.weights
												for layer_name, layer_module in attn_proc.named_modules():
																		if layer_name.endswith("""_lora"""      ):
																								snake_case_					:int		=     (
																								    torch.randn_like(layer_module.up.weight      )
																								    if randn_weight
																								    else torch.zeros_like(layer_module.up.weight      )
																								)
																								snake_case_					:str		=     torch.nn.Parameter(_lowercase      )

class 							lowerCamelCase    (     unittest.TestCase							):






					'''simple docstring'''





					def   lowerCAmelCase_							(			self:				Any		)  ->      Tuple:
											torch.manual_seed(0		)
											snake_case_					:List[Any]		=     UNetaDConditionModel(
											    block_out_channels=(32, 64)		,	layers_per_block=2		,	sample_size=32		,	in_channels=4		,	out_channels=4		,	down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")		,	up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")		,	cross_attention_dim=32		,	)
											snake_case_					:Optional[int]		=     DDIMScheduler(
											    beta_start=0.0_0_0_8_5		,	beta_end=0.0_1_2		,	beta_schedule="""scaled_linear"""		,	clip_sample=snake_case		,	set_alpha_to_one=snake_case		,	steps_offset=1		,	)
											torch.manual_seed(0		)
											snake_case_					:int		=     AutoencoderKL(
											    block_out_channels=[32, 64]		,	in_channels=3		,	out_channels=3		,	down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]		,	up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]		,	latent_channels=4		,	)
											snake_case_					:Union[str, Any]		=     CLIPTextConfig(
											    bos_token_id=0		,	eos_token_id=2		,	hidden_size=32		,	intermediate_size=37		,	layer_norm_eps=1E-05		,	num_attention_heads=4		,	num_hidden_layers=5		,	pad_token_id=1		,	vocab_size=1_000		,	)
											snake_case_					:Tuple		=     CLIPTextModel(snake_case		)
											snake_case_					:Optional[int]		=     CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""		)

											snake_case_,			snake_case_					:List[Any]		=     create_unet_lora_layers(snake_case		)
											snake_case_					:Tuple		=     create_text_encoder_lora_layers(snake_case		)

											snake_case_					:List[str]		=     {
											    """unet""": unet,
											    """scheduler""": scheduler,
											    """vae""": vae,
											    """text_encoder""": text_encoder,
											    """tokenizer""": tokenizer,
											    """safety_checker""": None,
											    """feature_extractor""": None,
											}
											snake_case_					:Union[str, Any]		=     {
											    """unet_lora_layers""": unet_lora_layers,
											    """text_encoder_lora_layers""": text_encoder_lora_layers,
											    """unet_lora_attn_procs""": unet_lora_attn_procs,
											}
											return pipeline_components, lora_components






					def   lowerCAmelCase_							(			self:				str		)  ->      int:
											snake_case_					:Tuple		=     1
											snake_case_					:Any		=     10
											snake_case_					:List[str]		=     4
											snake_case_					:Optional[Any]		=     (32, 32)

											snake_case_					:List[Any]		=     torch.manual_seed(0		)
											snake_case_					:List[Any]		=     floats_tensor((batch_size, num_channels) + sizes		)
											snake_case_					:Optional[int]		=     torch.randint(1		,	snake_case		,	size=(batch_size, sequence_length)		,	generator=snake_case		)

											snake_case_					:Union[str, Any]		=     {
											    """prompt""": """A painting of a squirrel eating a burger""",
											    """generator""": generator,
											    """num_inference_steps""": 2,
											    """guidance_scale""": 6.0,
											    """output_type""": """numpy""",
											}

											return noise, input_ids, pipeline_inputs

											# copied from: https://colab.research.google.com/gist/sayakpaul/df2ef6e1ae6d8c10a49d859883b10860/scratchpad.ipynb






					def   lowerCAmelCase_							(			self:				Optional[int]		)  ->      Optional[int]:
											snake_case_					:Tuple		=     77

											snake_case_					:str		=     torch.randint(2		,	56		,	size=(1, max_seq_length)		,	generator=torch.manual_seed(0		)		)

											snake_case_					:List[str]		=     {}
											snake_case_					:Optional[Any]		=     inputs
											return prepared_inputs






					def   lowerCAmelCase_							(			self:				Optional[Any]		,	snake_case:				Any		)  ->      List[str]:
											snake_case_,			snake_case_					:Union[str, Any]		=     self.get_dummy_components()
											LoraLoaderMixin.save_lora_weights(
											    save_directory=snake_case		,	unet_lora_layers=lora_components["""unet_lora_layers"""]		,	text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]		,	)
											self.assertTrue(os.path.isfile(os.path.join(snake_case		,	"""pytorch_lora_weights.bin"""		)		)		)






					def   lowerCAmelCase_							(			self:				Optional[Any]		)  ->      Optional[Any]:
											snake_case_,			snake_case_					:Optional[int]		=     self.get_dummy_components()
											snake_case_					:Optional[Any]		=     StableDiffusionPipeline(**snake_case		)
											snake_case_					:Tuple		=     sd_pipe.to(snake_case		)
											sd_pipe.set_progress_bar_config(disable=snake_case		)

											snake_case_,			snake_case_,			snake_case_					:Optional[int]		=     self.get_dummy_inputs()

											snake_case_					:Dict		=     sd_pipe(**snake_case		).images
											snake_case_					:Union[str, Any]		=     original_images[0, -3:, -3:, -1]

											with tempfile.TemporaryDirectory() as tmpdirname:
																	LoraLoaderMixin.save_lora_weights(
																	    save_directory=snake_case		,	unet_lora_layers=lora_components["""unet_lora_layers"""]		,	text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]		,	)
																	self.assertTrue(os.path.isfile(os.path.join(snake_case		,	"""pytorch_lora_weights.bin"""		)		)		)
																	sd_pipe.load_lora_weights(snake_case		)

											snake_case_					:Any		=     sd_pipe(**snake_case		).images
											snake_case_					:int		=     lora_images[0, -3:, -3:, -1]

											# Outputs shouldn't match.
											self.assertFalse(torch.allclose(torch.from_numpy(snake_case		)		,	torch.from_numpy(snake_case		)		)		)






					def   lowerCAmelCase_							(			self:				List[Any]		)  ->      Tuple:
											snake_case_,			snake_case_					:Optional[int]		=     self.get_dummy_components()
											snake_case_					:Tuple		=     StableDiffusionPipeline(**snake_case		)
											snake_case_					:Dict		=     sd_pipe.to(snake_case		)
											sd_pipe.set_progress_bar_config(disable=snake_case		)

											snake_case_,			snake_case_,			snake_case_					:Dict		=     self.get_dummy_inputs()

											snake_case_					:Optional[Any]		=     sd_pipe(**snake_case		).images
											snake_case_					:str		=     original_images[0, -3:, -3:, -1]

											with tempfile.TemporaryDirectory() as tmpdirname:
																	LoraLoaderMixin.save_lora_weights(
																	    save_directory=snake_case		,	unet_lora_layers=lora_components["""unet_lora_layers"""]		,	text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]		,	safe_serialization=snake_case		,	)
																	self.assertTrue(os.path.isfile(os.path.join(snake_case		,	"""pytorch_lora_weights.safetensors"""		)		)		)
																	sd_pipe.load_lora_weights(snake_case		)

											snake_case_					:Any		=     sd_pipe(**snake_case		).images
											snake_case_					:int		=     lora_images[0, -3:, -3:, -1]

											# Outputs shouldn't match.
											self.assertFalse(torch.allclose(torch.from_numpy(snake_case		)		,	torch.from_numpy(snake_case		)		)		)






					def   lowerCAmelCase_							(			self:				Optional[int]		)  ->      str:
											snake_case_,			snake_case_					:Union[str, Any]		=     self.get_dummy_components()
											snake_case_					:Optional[int]		=     lora_components["""unet_lora_attn_procs"""]
											snake_case_					:Dict		=     StableDiffusionPipeline(**snake_case		)
											snake_case_					:Optional[int]		=     sd_pipe.to(snake_case		)
											sd_pipe.set_progress_bar_config(disable=snake_case		)

											snake_case_,			snake_case_,			snake_case_					:str		=     self.get_dummy_inputs()

											snake_case_					:Optional[Any]		=     sd_pipe(**snake_case		).images
											snake_case_					:List[Any]		=     original_images[0, -3:, -3:, -1]

											with tempfile.TemporaryDirectory() as tmpdirname:
																	snake_case_					:List[str]		=     sd_pipe.unet
																	unet.set_attn_processor(snake_case		)
																	unet.save_attn_procs(snake_case		)
																	self.assertTrue(os.path.isfile(os.path.join(snake_case		,	"""pytorch_lora_weights.bin"""		)		)		)
																	sd_pipe.load_lora_weights(snake_case		)

											snake_case_					:str		=     sd_pipe(**snake_case		).images
											snake_case_					:Optional[Any]		=     lora_images[0, -3:, -3:, -1]

											# Outputs shouldn't match.
											self.assertFalse(torch.allclose(torch.from_numpy(snake_case		)		,	torch.from_numpy(snake_case		)		)		)






					def   lowerCAmelCase_							(			self:				Any		)  ->      Optional[Any]:
											snake_case_,			snake_case_					:int		=     self.get_dummy_components()
											snake_case_					:Union[str, Any]		=     StableDiffusionPipeline(**snake_case		)

											snake_case_					:Optional[int]		=     self.get_dummy_tokens()

											# inference without lora
											snake_case_					:Optional[Any]		=     pipe.text_encoder(**snake_case		)[0]
											assert outputs_without_lora.shape == (1, 77, 32)

											# create lora_attn_procs with zeroed out up.weights
											snake_case_					:Union[str, Any]		=     create_text_encoder_lora_attn_procs(pipe.text_encoder		)
											set_lora_up_weights(snake_case		,	randn_weight=snake_case		)

											# monkey patch
											pipe._modify_text_encoder(snake_case		)

											# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
											del text_attn_procs
											gc.collect()

											# inference with lora
											snake_case_					:Any		=     pipe.text_encoder(**snake_case		)[0]
											assert outputs_with_lora.shape == (1, 77, 32)

											assert torch.allclose(
											    snake_case		,	snake_case		), "lora_up_weight are all zero, so the lora outputs should be the same to without lora outputs"

											# create lora_attn_procs with randn up.weights
											snake_case_					:Dict		=     create_text_encoder_lora_attn_procs(pipe.text_encoder		)
											set_lora_up_weights(snake_case		,	randn_weight=snake_case		)

											# monkey patch
											pipe._modify_text_encoder(snake_case		)

											# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
											del text_attn_procs
											gc.collect()

											# inference with lora
											snake_case_					:Dict		=     pipe.text_encoder(**snake_case		)[0]
											assert outputs_with_lora.shape == (1, 77, 32)

											assert not torch.allclose(
											    snake_case		,	snake_case		), "lora_up_weight are not zero, so the lora outputs should be different to without lora outputs"






					def   lowerCAmelCase_							(			self:				Optional[int]		)  ->      Optional[Any]:
											snake_case_,			snake_case_					:Optional[Any]		=     self.get_dummy_components()
											snake_case_					:List[Any]		=     StableDiffusionPipeline(**snake_case		)

											snake_case_					:Tuple		=     self.get_dummy_tokens()

											# inference without lora
											snake_case_					:Union[str, Any]		=     pipe.text_encoder(**snake_case		)[0]
											assert outputs_without_lora.shape == (1, 77, 32)

											# create lora_attn_procs with randn up.weights
											snake_case_					:List[str]		=     create_text_encoder_lora_attn_procs(pipe.text_encoder		)
											set_lora_up_weights(snake_case		,	randn_weight=snake_case		)

											# monkey patch
											pipe._modify_text_encoder(snake_case		)

											# verify that it's okay to release the text_attn_procs which holds the LoRAAttnProcessor.
											del text_attn_procs
											gc.collect()

											# inference with lora
											snake_case_					:Tuple		=     pipe.text_encoder(**snake_case		)[0]
											assert outputs_with_lora.shape == (1, 77, 32)

											assert not torch.allclose(
											    snake_case		,	snake_case		), "lora outputs should be different to without lora outputs"

											# remove monkey patch
											pipe._remove_text_encoder_monkey_patch()

											# inference with removed lora
											snake_case_					:str		=     pipe.text_encoder(**snake_case		)[0]
											assert outputs_without_lora_removed.shape == (1, 77, 32)

											assert torch.allclose(
											    snake_case		,	snake_case		), "remove lora monkey patch should restore the original outputs"






					def   lowerCAmelCase_							(			self:				Optional[Any]		)  ->      List[str]:
											snake_case_,			snake_case_					:List[Any]		=     self.get_dummy_components()
											snake_case_					:str		=     StableDiffusionPipeline(**snake_case		)
											snake_case_					:Tuple		=     sd_pipe.to(snake_case		)
											sd_pipe.set_progress_bar_config(disable=snake_case		)

											snake_case_,			snake_case_,			snake_case_					:List[str]		=     self.get_dummy_inputs()

											with tempfile.TemporaryDirectory() as tmpdirname:
																	LoraLoaderMixin.save_lora_weights(
																	    save_directory=snake_case		,	unet_lora_layers=lora_components["""unet_lora_layers"""]		,	text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]		,	)
																	self.assertTrue(os.path.isfile(os.path.join(snake_case		,	"""pytorch_lora_weights.bin"""		)		)		)
																	sd_pipe.load_lora_weights(snake_case		)

											snake_case_					:Optional[Any]		=     sd_pipe(**snake_case		).images
											snake_case_					:List[str]		=     lora_images[0, -3:, -3:, -1]

											snake_case_					:Optional[Any]		=     sd_pipe(**snake_case		,	cross_attention_kwargs={"""scale""": 0.5}		).images
											snake_case_					:int		=     lora_images_with_scale[0, -3:, -3:, -1]

											# Outputs shouldn't match.
											self.assertFalse(
											    torch.allclose(torch.from_numpy(snake_case		)		,	torch.from_numpy(snake_case		)		)		)






					def   lowerCAmelCase_							(			self:				Tuple		)  ->      Any:
											with tempfile.TemporaryDirectory() as tmpdirname:
																	self.create_lora_weight_file(snake_case		)

																	snake_case_,			snake_case_					:int		=     self.get_dummy_components()
																	snake_case_					:Tuple		=     StableDiffusionPipeline(**snake_case		)
																	snake_case_					:List[str]		=     sd_pipe.to(snake_case		)
																	sd_pipe.set_progress_bar_config(disable=snake_case		)

																	# check if vanilla attention processors are used
																	for _, module in sd_pipe.unet.named_modules():
																							if isinstance(snake_case		,	snake_case		):
																													self.assertIsInstance(module.processor		,	(AttnProcessor, AttnProcessor2_0)		)

            # load LoRA weight file
																	sd_pipe.load_lora_weights(snake_case		)

																	# check if lora attention processors are used
																	for _, module in sd_pipe.unet.named_modules():
																							if isinstance(snake_case		,	snake_case		):
																													snake_case_					:Dict		=     (
																													    LoRAAttnProcessor2_0 if hasattr(snake_case		,	"""scaled_dot_product_attention"""		) else LoRAAttnProcessor
																													)
																													self.assertIsInstance(module.processor		,	snake_case		)






					@unittest.skipIf(torch_device != """cuda"""		,	"""This test is supposed to run on GPU"""		)
					def   lowerCAmelCase_							(			self:				Optional[Any]		)  ->      List[str]:
											with tempfile.TemporaryDirectory() as tmpdirname:
																	self.create_lora_weight_file(snake_case		)

																	snake_case_,			snake_case_					:Union[str, Any]		=     self.get_dummy_components()
																	snake_case_					:Optional[Any]		=     StableDiffusionPipeline(**snake_case		)
																	snake_case_					:int		=     sd_pipe.to(snake_case		)
																	sd_pipe.set_progress_bar_config(disable=snake_case		)

																	# enable XFormers
																	sd_pipe.enable_xformers_memory_efficient_attention()

																	# check if xFormers attention processors are used
																	for _, module in sd_pipe.unet.named_modules():
																							if isinstance(snake_case		,	snake_case		):
																													self.assertIsInstance(module.processor		,	snake_case		)

            # load LoRA weight file
																	sd_pipe.load_lora_weights(snake_case		)

																	# check if lora attention processors are used
																	for _, module in sd_pipe.unet.named_modules():
																							if isinstance(snake_case		,	snake_case		):
																													self.assertIsInstance(module.processor		,	snake_case		)






					@unittest.skipIf(torch_device != """cuda"""		,	"""This test is supposed to run on GPU"""		)
					def   lowerCAmelCase_							(			self:				Union[str, Any]		)  ->      int:
											snake_case_,			snake_case_					:Dict		=     self.get_dummy_components()
											snake_case_					:Optional[int]		=     StableDiffusionPipeline(**snake_case		)
											snake_case_					:List[Any]		=     sd_pipe.to(snake_case		)
											sd_pipe.set_progress_bar_config(disable=snake_case		)

											snake_case_,			snake_case_,			snake_case_					:List[str]		=     self.get_dummy_inputs()

											# enable XFormers
											sd_pipe.enable_xformers_memory_efficient_attention()

											snake_case_					:str		=     sd_pipe(**snake_case		).images
											snake_case_					:Dict		=     original_images[0, -3:, -3:, -1]

											with tempfile.TemporaryDirectory() as tmpdirname:
																	LoraLoaderMixin.save_lora_weights(
																	    save_directory=snake_case		,	unet_lora_layers=lora_components["""unet_lora_layers"""]		,	text_encoder_lora_layers=lora_components["""text_encoder_lora_layers"""]		,	)
																	self.assertTrue(os.path.isfile(os.path.join(snake_case		,	"""pytorch_lora_weights.bin"""		)		)		)
																	sd_pipe.load_lora_weights(snake_case		)

											snake_case_					:List[str]		=     sd_pipe(**snake_case		).images
											snake_case_					:Dict		=     lora_images[0, -3:, -3:, -1]

											# Outputs shouldn't match.
											self.assertFalse(torch.allclose(torch.from_numpy(snake_case		)		,	torch.from_numpy(snake_case		)		)		)





"""simple docstring"""
import inspect
import os
from argparse import ArgumentParser, Namespace
from importlib import import_module

import huggingface_hub
import numpy as np
from packaging import version

from .. import (
    FEATURE_EXTRACTOR_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    PROCESSOR_MAPPING,
    TOKENIZER_MAPPING,
    AutoConfig,
    AutoFeatureExtractor,
    AutoImageProcessor,
    AutoProcessor,
    AutoTokenizer,
    is_datasets_available,
    is_tf_available,
    is_torch_available,
)
from ..utils import TF2_WEIGHTS_INDEX_NAME, TF2_WEIGHTS_NAME, logging
from . import BaseTransformersCLICommand


if is_tf_available():
						import tensorflow as tf

						tf.config.experimental.enable_tensor_float_aa_execution(False)

if is_torch_available():
						import torch

if is_datasets_available():
						from datasets import load_dataset


__a           =							5e-5  # larger error tolerance than in our internal tests, to avoid flaky user-facing errors
def   A_    (     _lowercase      ):







						'''simple docstring'''


						return PTtoTFCommand(
						    args.model_name,   args.local_dir,   args.max_error,   args.new_weights,   args.no_pr,   args.push,   args.extra_commit_description,   args.override_model_class,   )

class 							lowerCamelCase    (     _lowerCAmelCase							):






					'''simple docstring'''





					@staticmethod
					def   lowerCAmelCase_							(			snake_case:				ArgumentParser		)  ->      Dict:
											snake_case_					:int		=     parser.add_parser(
											    """pt-to-tf"""		,	help=(
											        """CLI tool to run convert a transformers model from a PyTorch checkpoint to a TensorFlow checkpoint."""
											        """ Can also be used to validate existing weights without opening PRs, with --no-pr."""
											    )		,	)
											train_parser.add_argument(
											    """--model-name"""		,	type=snake_case		,	required=snake_case		,	help="""The model name, including owner/organization, as seen on the hub."""		,	)
											train_parser.add_argument(
											    """--local-dir"""		,	type=snake_case		,	default=""""""		,	help="""Optional local directory of the model repository. Defaults to /tmp/{model_name}"""		,	)
											train_parser.add_argument(
											    """--max-error"""		,	type=snake_case		,	default=snake_case		,	help=(
											        f"""Maximum error tolerance. Defaults to {MAX_ERROR}. This flag should be avoided, use at your own risk."""
											    )		,	)
											train_parser.add_argument(
											    """--new-weights"""		,	action="""store_true"""		,	help="""Optional flag to create new TensorFlow weights, even if they already exist."""		,	)
											train_parser.add_argument(
											    """--no-pr"""		,	action="""store_true"""		,	help="""Optional flag to NOT open a PR with converted weights."""		)
											train_parser.add_argument(
											    """--push"""		,	action="""store_true"""		,	help="""Optional flag to push the weights directly to `main` (requires permissions)"""		,	)
											train_parser.add_argument(
											    """--extra-commit-description"""		,	type=snake_case		,	default=""""""		,	help="""Optional additional commit description to use when opening a PR (e.g. to tag the owner)."""		,	)
											train_parser.add_argument(
											    """--override-model-class"""		,	type=snake_case		,	default=snake_case		,	help="""If you think you know better than the auto-detector, you can specify the model class here. """
											    """Can be either an AutoModel class or a specific model class like BertForSequenceClassification."""		,	)
											train_parser.set_defaults(func=snake_case		)






					@staticmethod
					def   lowerCAmelCase_							(			snake_case:				Union[str, Any]		,	snake_case:				List[str]		)  ->      List[str]:
											snake_case_					:List[Any]		=     set(pt_outputs.keys()		)
											snake_case_					:Optional[Any]		=     set(tf_outputs.keys()		)
											if pt_out_attrs != tf_out_attrs:
																	raise ValueError(
																	    f"""The model outputs have different attributes, aborting. (Pytorch: {pt_out_attrs}, TensorFlow:"""
																	    f""" {tf_out_attrs})"""		)

											# 2. For each output attribute, computes the difference
											def _find_pt_tf_differences(snake_case:				List[Any]		,	snake_case:				List[Any]		,	snake_case:				Optional[int]		,	snake_case:				int=""		):
																	# If the current attribute is a tensor, it is a leaf and we make the comparison. Otherwise, we will dig in
																	# recursivelly, keeping the name of the attribute.
																	if isinstance(snake_case		,	torch.Tensor		):
																							snake_case_					:List[str]		=     np.max(np.abs(pt_out.numpy() - tf_out.numpy()		)		)
																							snake_case_					:Optional[int]		=     tensor_difference
																	else:
																							snake_case_					:Optional[int]		=     attr_name
																							for i, pt_item in enumerate(snake_case		):
																													# If it is a named attribute, we keep the name. Otherwise, just its index.
																													if isinstance(snake_case		,	snake_case		):
																																			snake_case_					:int		=     root_name + pt_item
																																			snake_case_					:Dict		=     tf_out[pt_item]
																																			snake_case_					:List[str]		=     pt_out[pt_item]
																													else:
																																			snake_case_					:str		=     root_name + f"""[{i}]"""
																																			snake_case_					:List[Any]		=     tf_out[i]
																													snake_case_					:Dict		=     _find_pt_tf_differences(snake_case		,	snake_case		,	snake_case		,	snake_case		)

																	return differences

											return _find_pt_tf_differences(snake_case		,	snake_case		,	{}		)






					def __init__(			self:				Tuple		,	snake_case:				str		,	snake_case:				str		,	snake_case:				float		,	snake_case:				bool		,	snake_case:				bool		,	snake_case:				bool		,	snake_case:				str		,	snake_case:				str		,	*snake_case:				Any		,	)  ->      Union[str, Any]:
											snake_case_					:List[Any]		=     logging.get_logger("""transformers-cli/pt_to_tf"""		)
											snake_case_					:Optional[int]		=     model_name
											snake_case_					:List[str]		=     local_dir if local_dir else os.path.join("""/tmp"""		,	snake_case		)
											snake_case_					:Optional[Any]		=     max_error
											snake_case_					:List[str]		=     new_weights
											snake_case_					:List[str]		=     no_pr
											snake_case_					:Dict		=     push
											snake_case_					:Tuple		=     extra_commit_description
											snake_case_					:Optional[Any]		=     override_model_class






					def   lowerCAmelCase_							(			self:				Tuple		,	snake_case:				List[Any]		,	snake_case:				str		,	snake_case:				Optional[Any]		)  ->      Optional[int]:

											def _get_audio_input():
																	snake_case_					:int		=     load_dataset("""hf-internal-testing/librispeech_asr_dummy"""		,	"""clean"""		,	split="""validation"""		)
																	snake_case_					:int		=     ds.sort("""id"""		).select(range(2		)		)[:2]["""audio"""]
																	snake_case_					:int		=     [x["""array"""] for x in speech_samples]
																	return raw_samples

											snake_case_					:List[str]		=     type(pt_model.config		)
											if model_config_class in PROCESSOR_MAPPING:
																	snake_case_					:Any		=     AutoProcessor.from_pretrained(self._local_dir		)
																	if model_config_class in TOKENIZER_MAPPING and processor.tokenizer.pad_token is None:
																							snake_case_					:List[str]		=     processor.tokenizer.eos_token
											elif model_config_class in IMAGE_PROCESSOR_MAPPING:
																	snake_case_					:List[Any]		=     AutoImageProcessor.from_pretrained(self._local_dir		)
											elif model_config_class in FEATURE_EXTRACTOR_MAPPING:
																	snake_case_					:int		=     AutoFeatureExtractor.from_pretrained(self._local_dir		)
											elif model_config_class in TOKENIZER_MAPPING:
																	snake_case_					:Tuple		=     AutoTokenizer.from_pretrained(self._local_dir		)
																	if processor.pad_token is None:
																							snake_case_					:str		=     processor.eos_token
											else:
																	raise ValueError(f"""Unknown data processing type (model config type: {model_config_class})"""		)

											snake_case_					:Optional[int]		=     set(inspect.signature(pt_model.forward		).parameters.keys()		)
											snake_case_					:Dict		=     {}
											if "input_ids" in model_forward_signature:
																	processor_inputs.update(
																	    {
																	        """text""": ["""Hi there!""", """I am a batch with more than one row and different input lengths."""],
																	        """padding""": True,
																	        """truncation""": True,
																	    }		)
											if "pixel_values" in model_forward_signature:
																	snake_case_					:Optional[int]		=     load_dataset("""cifar10"""		,	"""plain_text"""		,	split="""test"""		)[:2]["""img"""]
																	processor_inputs.update({"""images""": sample_images}		)
											if "input_features" in model_forward_signature:
																	snake_case_					:List[str]		=     inspect.signature(processor.feature_extractor		).parameters
																	# Pad to the largest input length by default but take feature extractor default
																	# padding value if it exists e.g. "max_length" and is not False or None
																	if "padding" in feature_extractor_signature:
																							snake_case_					:Tuple		=     feature_extractor_signature["""padding"""].default
																							if default_strategy is not False and default_strategy is not None:
																													snake_case_					:Any		=     default_strategy
																							else:
																													snake_case_					:Optional[Any]		=     True
																	else:
																							snake_case_					:List[Any]		=     True
																	processor_inputs.update({"""audio""": _get_audio_input(), """padding""": padding_strategy}		)
											if "input_values" in model_forward_signature:  # Wav2Vec2 audio input
																	processor_inputs.update({"""audio""": _get_audio_input(), """padding""": True}		)
											snake_case_					:Any		=     processor(**snake_case		,	return_tensors="""pt"""		)
											snake_case_					:Union[str, Any]		=     processor(**snake_case		,	return_tensors="""tf"""		)

											# Extra input requirements, in addition to the input modality
											if (
											    config.is_encoder_decoder
											    or (hasattr(snake_case		,	"""encoder"""		) and hasattr(snake_case		,	"""decoder"""		))
											    or "decoder_input_ids" in tf_dummy_inputs
											):
																	snake_case_					:Optional[int]		=     np.asarray([[1], [1]]		,	dtype=snake_case		) * (pt_model.config.decoder_start_token_id or 0)
																	pt_input.update({"""decoder_input_ids""": torch.tensor(snake_case		)}		)
																	tf_input.update({"""decoder_input_ids""": tf.convert_to_tensor(snake_case		)}		)

											return pt_input, tf_input






					def   lowerCAmelCase_							(			self:				Tuple		)  ->      List[str]:
											# hub version 0.9.0 introduced the possibility of programmatically opening PRs with normal write tokens.
											if version.parse(huggingface_hub.__version__		) < version.parse("""0.9.0"""		):
																	raise ImportError(
																	    """The huggingface_hub version must be >= 0.9.0 to use this command. Please update your huggingface_hub"""
																	    """ installation."""		)
											else:
																	from huggingface_hub import Repository, create_commit
																	from huggingface_hub._commit_api import CommitOperationAdd

											# Fetch remote data
											snake_case_					:Tuple		=     Repository(local_dir=self._local_dir		,	clone_from=self._model_name		)

											# Load config and get the appropriate architecture -- the latter is needed to convert the head's weights
											snake_case_					:Dict		=     AutoConfig.from_pretrained(self._local_dir		)
											snake_case_					:Union[str, Any]		=     config.architectures
											if self._override_model_class is not None:
																	if self._override_model_class.startswith("""TF"""		):
																							snake_case_					:int		=     [self._override_model_class[2:]]
																	else:
																							snake_case_					:Tuple		=     [self._override_model_class]
																	try:
																							snake_case_					:Optional[Any]		=     getattr(import_module("""transformers"""		)		,	architectures[0]		)
																	except AttributeError:
																							raise ValueError(f"""Model class {self._override_model_class} not found in transformers."""		)
																	try:
																							snake_case_					:List[str]		=     getattr(import_module("""transformers"""		)		,	"""TF""" + architectures[0]		)
																	except AttributeError:
																							raise ValueError(f"""TF model class TF{self._override_model_class} not found in transformers."""		)
											elif architectures is None:  # No architecture defined -- use auto classes
																	snake_case_					:Any		=     getattr(import_module("""transformers"""		)		,	"""AutoModel"""		)
																	snake_case_					:int		=     getattr(import_module("""transformers"""		)		,	"""TFAutoModel"""		)
																	self._logger.warning("""No detected architecture, using AutoModel/TFAutoModel"""		)
											else:  # Architecture defined -- use it
																	if len(snake_case		) > 1:
																							raise ValueError(f"""More than one architecture was found, aborting. (architectures = {architectures})"""		)
																	self._logger.warning(f"""Detected architecture: {architectures[0]}"""		)
																	snake_case_					:str		=     getattr(import_module("""transformers"""		)		,	architectures[0]		)
																	try:
																							snake_case_					:Optional[Any]		=     getattr(import_module("""transformers"""		)		,	"""TF""" + architectures[0]		)
																	except AttributeError:
																							raise AttributeError(f"""The TensorFlow equivalent of {architectures[0]} doesn't exist in transformers."""		)

        # Check the TF dummy inputs to see what keys we need in the forward pass
											snake_case_					:List[str]		=     tf_class.from_config(snake_case		)
											snake_case_					:str		=     tf_from_pt_model.dummy_inputs

											del tf_from_pt_model  # Try to keep only one model in memory at a time

											# Load the model and get some basic inputs
											snake_case_					:Any		=     pt_class.from_pretrained(self._local_dir		)
											pt_model.eval()

											snake_case_,			snake_case_					:List[Any]		=     self.get_inputs(snake_case		,	snake_case		,	snake_case		)

											with torch.no_grad():
																	snake_case_					:str		=     pt_model(**snake_case		,	output_hidden_states=snake_case		)
											del pt_model  # will no longer be used, and may have a large memory footprint

											snake_case_					:Any		=     tf_class.from_pretrained(self._local_dir		,	from_pt=snake_case		)
											snake_case_					:Dict		=     tf_from_pt_model(**snake_case		,	output_hidden_states=snake_case		,	training=snake_case		)

											# Confirms that cross loading PT weights into TF worked.
											snake_case_					:Optional[Any]		=     self.find_pt_tf_differences(snake_case		,	snake_case		)
											snake_case_					:int		=     {k: v for k, v in crossload_differences.items() if """hidden""" not in k}
											snake_case_					:List[Any]		=     {k: v for k, v in crossload_differences.items() if """hidden""" in k}
											if len(snake_case		) == 0 and architectures is not None:
																	raise ValueError(
																	    f"""Something went wrong -- the config file has architectures ({architectures}), but no model head"""
																	    """ output was found. All outputs start with 'hidden'"""		)
											snake_case_					:Any		=     max(output_differences.values()		) if output_differences else 0.0
											snake_case_					:List[str]		=     max(hidden_differences.values()		)
											if max_crossload_output_diff > self._max_error or max_crossload_hidden_diff > self._max_error:
																	raise ValueError(
																	    """The cross-loaded TensorFlow model has different outputs, something went wrong!\n"""
																	    + f"""\nList of maximum output differences above the threshold ({self._max_error}):\n"""
																	    + """\n""".join([f"""{k}: {v:.3e}""" for k, v in output_differences.items() if v > self._max_error]		)
																	    + f"""\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n"""
																	    + """\n""".join([f"""{k}: {v:.3e}""" for k, v in hidden_differences.items() if v > self._max_error]		)		)

											# Save the weights in a TF format (if needed) and confirms that the results are still good
											snake_case_					:Tuple		=     os.path.join(self._local_dir		,	snake_case		)
											snake_case_					:Union[str, Any]		=     os.path.join(self._local_dir		,	snake_case		)
											if (not os.path.exists(snake_case		) and not os.path.exists(snake_case		)) or self._new_weights:
																	tf_from_pt_model.save_pretrained(self._local_dir		)
											del tf_from_pt_model  # will no longer be used, and may have a large memory footprint

											snake_case_					:Optional[Any]		=     tf_class.from_pretrained(self._local_dir		)
											snake_case_					:Dict		=     tf_model(**snake_case		,	output_hidden_states=snake_case		)

											snake_case_					:List[Any]		=     self.find_pt_tf_differences(snake_case		,	snake_case		)
											snake_case_					:Optional[int]		=     {k: v for k, v in conversion_differences.items() if """hidden""" not in k}
											snake_case_					:List[Any]		=     {k: v for k, v in conversion_differences.items() if """hidden""" in k}
											if len(snake_case		) == 0 and architectures is not None:
																	raise ValueError(
																	    f"""Something went wrong -- the config file has architectures ({architectures}), but no model head"""
																	    """ output was found. All outputs start with 'hidden'"""		)
											snake_case_					:Dict		=     max(output_differences.values()		) if output_differences else 0.0
											snake_case_					:Tuple		=     max(hidden_differences.values()		)
											if max_conversion_output_diff > self._max_error or max_conversion_hidden_diff > self._max_error:
																	raise ValueError(
																	    """The converted TensorFlow model has different outputs, something went wrong!\n"""
																	    + f"""\nList of maximum output differences above the threshold ({self._max_error}):\n"""
																	    + """\n""".join([f"""{k}: {v:.3e}""" for k, v in output_differences.items() if v > self._max_error]		)
																	    + f"""\n\nList of maximum hidden layer differences above the threshold ({self._max_error}):\n"""
																	    + """\n""".join([f"""{k}: {v:.3e}""" for k, v in hidden_differences.items() if v > self._max_error]		)		)

											snake_case_					:Tuple		=     """Update TF weights""" if self._new_weights else """Add TF weights"""
											if self._push:
																	repo.git_add(auto_lfs_track=snake_case		)
																	repo.git_commit(snake_case		)
																	repo.git_push(blocking=snake_case		)  # this prints a progress bar with the upload
																	self._logger.warning(f"""TF weights pushed into {self._model_name}"""		)
											elif not self._no_pr:
																	self._logger.warning("""Uploading the weights into a new PR..."""		)
																	snake_case_					:Tuple		=     (
																	    """Model converted by the [`transformers`' `pt_to_tf`"""
																	    """ CLI](https://github.com/huggingface/transformers/blob/main/src/transformers/commands/pt_to_tf.py). """
																	    """All converted model outputs and hidden layers were validated against its PyTorch counterpart.\n\n"""
																	    f"""Maximum crossload output difference={max_crossload_output_diff:.3e}; """
																	    f"""Maximum crossload hidden layer difference={max_crossload_hidden_diff:.3e};\n"""
																	    f"""Maximum conversion output difference={max_conversion_output_diff:.3e}; """
																	    f"""Maximum conversion hidden layer difference={max_conversion_hidden_diff:.3e};\n"""
																	)
																	if self._max_error > MAX_ERROR:
																							commit_descrition += (
																							    f"""\n\nCAUTION: The maximum admissible error was manually increased to {self._max_error}!"""
																							)
																	if self._extra_commit_description:
																							commit_descrition += "\n\n" + self._extra_commit_description

																	# sharded model -> adds all related files (index and .h5 shards)
																	if os.path.exists(snake_case		):
																							snake_case_					:str		=     [
																							    CommitOperationAdd(path_in_repo=snake_case		,	path_or_fileobj=snake_case		)
																							]
																							for shard_path in tf.io.gfile.glob(self._local_dir + """/tf_model-*.h5"""		):
																													operations += [
																													    CommitOperationAdd(path_in_repo=os.path.basename(snake_case		)		,	path_or_fileobj=snake_case		)
																													]
																	else:
																							snake_case_					:Optional[Any]		=     [CommitOperationAdd(path_in_repo=snake_case		,	path_or_fileobj=snake_case		)]

																	snake_case_					:Tuple		=     create_commit(
																	    repo_id=self._model_name		,	operations=snake_case		,	commit_message=snake_case		,	commit_description=snake_case		,	repo_type="""model"""		,	create_pr=snake_case		,	).pr_url
																	self._logger.warning(f"""PR open in {hub_pr_url}"""		)
