





from typing import Optional, Tuple, Union

import tensorflow as tf

from ...activations_tf import ACTaFN
from ...modeling_tf_outputs import (
    TFBaseModelOutputWithNoAttention,
    TFBaseModelOutputWithPoolingAndNoAttention,
    TFImageClassifierOutputWithNoAttention,
)
from ...modeling_tf_utils import TFPreTrainedModel, TFSequenceClassificationLoss, keras_serializable, unpack_inputs
from ...tf_utils import shape_list
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging
from .configuration_resnet import ResNetConfig


SCREAMING_SNAKE_CASE_		  =			logging.get_logger(__name__)

# General docstring
SCREAMING_SNAKE_CASE_		  =			'ResNetConfig'

# Base docstring
SCREAMING_SNAKE_CASE_		  =			'microsoft/resnet-50'
SCREAMING_SNAKE_CASE_		  =			[1, 2_0_4_8, 7, 7]

# Image classification docstring
SCREAMING_SNAKE_CASE_		  =			'microsoft/resnet-50'
SCREAMING_SNAKE_CASE_		  =			'tiger cat'

SCREAMING_SNAKE_CASE_		  =			[
    'microsoft/resnet-50',
    # See all resnet models at https://huggingface.co/models?filter=resnet
]

class 						UpperCamelCase__					( tf.keras.layers.Layer  ):



     '''simple docstring'''

     def __init__(				self		:       Union[str, Any]		,lowerCamelCase__		:       int		,lowerCamelCase__		:       int = 3		,lowerCamelCase__		:       int = 1		,lowerCamelCase__		:       str = "relu"		,**lowerCamelCase__		:       Union[str, Any]			)	->						int:




           '''simple docstring'''


           super().__init__(**_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		kernel_size // 2
           SCREAMING_SNAKE_CASE						=		tf.keras.layers.ConvaD(
               _lowerCamelCase		,kernel_size=_lowerCamelCase		,strides=_lowerCamelCase		,padding="""valid"""		,use_bias=_lowerCamelCase		,name="""convolution"""			)
           # Use same default momentum and epsilon as PyTorch equivalent
           SCREAMING_SNAKE_CASE						=		tf.keras.layers.BatchNormalization(epsilon=1e-5		,momentum=0.9		,name="""normalization"""			)
           SCREAMING_SNAKE_CASE						=		ACTaFN[activation] if activation is not None else tf.keras.layers.Activation("""linear"""			)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[str]		,lowerCamelCase__		:       tf.Tensor			)	->						Dict:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		(self.pad_value, self.pad_value)
           SCREAMING_SNAKE_CASE						=		tf.pad(_lowerCamelCase		,[(0, 0), height_pad, width_pad, (0, 0)]			)
           SCREAMING_SNAKE_CASE						=		self.conv(_lowerCamelCase			)
           return hidden_state






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Optional[Any]		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       bool = False			)	->						List[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		self.convolution(_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.normalization(_lowerCamelCase		,training=_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.activation(_lowerCamelCase			)
           return hidden_state

class 						UpperCamelCase__					( tf.keras.layers.Layer  ):



     '''simple docstring'''

     def __init__(				self		:       Dict		,lowerCamelCase__		:       ResNetConfig		,**lowerCamelCase__		:       List[Any]			)	->						Any:




           '''simple docstring'''


           super().__init__(**_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		TFResNetConvLayer(
               config.embedding_size		,kernel_size=7		,stride=2		,activation=config.hidden_act		,name="""embedder"""		,)
           SCREAMING_SNAKE_CASE						=		tf.keras.layers.MaxPoolaD(pool_size=3		,strides=2		,padding="""valid"""		,name="""pooler"""			)
           SCREAMING_SNAKE_CASE						=		config.num_channels






     def 							SCREAMING_SNAKE_CASE__    (				self		:       str		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       bool = False			)	->						Any:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		shape_list(_lowerCamelCase			)
           if tf.executing_eagerly() and num_channels != self.num_channels:
                 raise ValueError(
                     """Make sure that the channel dimension of the pixel values match with the one set in the configuration."""			)
           SCREAMING_SNAKE_CASE						=		pixel_values
           SCREAMING_SNAKE_CASE						=		self.embedder(_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		tf.pad(_lowerCamelCase		,[[0, 0], [1, 1], [1, 1], [0, 0]]			)
           SCREAMING_SNAKE_CASE						=		self.pooler(_lowerCamelCase			)
           return hidden_state

class 						UpperCamelCase__					( tf.keras.layers.Layer  ):



     '''simple docstring'''

     def __init__(				self		:       int		,lowerCamelCase__		:       int		,lowerCamelCase__		:       int = 2		,**lowerCamelCase__		:       int			)	->						List[str]:




           '''simple docstring'''


           super().__init__(**_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		tf.keras.layers.ConvaD(
               _lowerCamelCase		,kernel_size=1		,strides=_lowerCamelCase		,use_bias=_lowerCamelCase		,name="""convolution"""			)
           # Use same default momentum and epsilon as PyTorch equivalent
           SCREAMING_SNAKE_CASE						=		tf.keras.layers.BatchNormalization(epsilon=1e-5		,momentum=0.9		,name="""normalization"""			)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[str]		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       bool = False			)	->						Optional[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		x
           SCREAMING_SNAKE_CASE						=		self.convolution(_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.normalization(_lowerCamelCase		,training=_lowerCamelCase			)
           return hidden_state

class 						UpperCamelCase__					( tf.keras.layers.Layer  ):



     '''simple docstring'''

     def __init__(				self		:       int		,lowerCamelCase__		:       int		,lowerCamelCase__		:       int		,lowerCamelCase__		:       int = 1		,lowerCamelCase__		:       str = "relu"		,**lowerCamelCase__		:       Optional[Any]			)	->						Optional[int]:




           '''simple docstring'''


           super().__init__(**_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		in_channels != out_channels or stride != 1
           SCREAMING_SNAKE_CASE						=		TFResNetConvLayer(_lowerCamelCase		,stride=_lowerCamelCase		,name="""layer.0"""			)
           SCREAMING_SNAKE_CASE						=		TFResNetConvLayer(_lowerCamelCase		,activation=_lowerCamelCase		,name="""layer.1"""			)
           SCREAMING_SNAKE_CASE						=		(
               TFResNetShortCut(_lowerCamelCase		,stride=_lowerCamelCase		,name="""shortcut"""			)
               if should_apply_shortcut
               else tf.keras.layers.Activation("""linear"""		,name="""shortcut"""			)
           )
           SCREAMING_SNAKE_CASE						=		ACTaFN[activation]






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Optional[Any]		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       bool = False			)	->						Tuple:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		hidden_state
           SCREAMING_SNAKE_CASE						=		self.conva(_lowerCamelCase		,training=_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.conva(_lowerCamelCase		,training=_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.shortcut(_lowerCamelCase		,training=_lowerCamelCase			)
           hidden_state += residual
           SCREAMING_SNAKE_CASE						=		self.activation(_lowerCamelCase			)
           return hidden_state

class 						UpperCamelCase__					( tf.keras.layers.Layer  ):



     '''simple docstring'''

     def __init__(				self		:       Any		,lowerCamelCase__		:       int		,lowerCamelCase__		:       int		,lowerCamelCase__		:       int = 1		,lowerCamelCase__		:       str = "relu"		,lowerCamelCase__		:       int = 4		,**lowerCamelCase__		:       Optional[int]		,)	->						int:




           '''simple docstring'''


           super().__init__(**_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		in_channels != out_channels or stride != 1
           SCREAMING_SNAKE_CASE						=		out_channels // reduction
           SCREAMING_SNAKE_CASE						=		TFResNetConvLayer(_lowerCamelCase		,kernel_size=1		,name="""layer.0"""			)
           SCREAMING_SNAKE_CASE						=		TFResNetConvLayer(_lowerCamelCase		,stride=_lowerCamelCase		,name="""layer.1"""			)
           SCREAMING_SNAKE_CASE						=		TFResNetConvLayer(_lowerCamelCase		,kernel_size=1		,activation=_lowerCamelCase		,name="""layer.2"""			)
           SCREAMING_SNAKE_CASE						=		(
               TFResNetShortCut(_lowerCamelCase		,stride=_lowerCamelCase		,name="""shortcut"""			)
               if should_apply_shortcut
               else tf.keras.layers.Activation("""linear"""		,name="""shortcut"""			)
           )
           SCREAMING_SNAKE_CASE						=		ACTaFN[activation]






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Dict		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       bool = False			)	->						Union[str, Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		hidden_state
           SCREAMING_SNAKE_CASE						=		self.conva(_lowerCamelCase		,training=_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.conva(_lowerCamelCase		,training=_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.conva(_lowerCamelCase		,training=_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.shortcut(_lowerCamelCase		,training=_lowerCamelCase			)
           hidden_state += residual
           SCREAMING_SNAKE_CASE						=		self.activation(_lowerCamelCase			)
           return hidden_state

class 						UpperCamelCase__					( tf.keras.layers.Layer  ):



     '''simple docstring'''

     def __init__(				self		:       Dict		,lowerCamelCase__		:       ResNetConfig		,lowerCamelCase__		:       int		,lowerCamelCase__		:       int		,lowerCamelCase__		:       int = 2		,lowerCamelCase__		:       int = 2		,**lowerCamelCase__		:       int			)	->						str:




           '''simple docstring'''


           super().__init__(**_lowerCamelCase			)

           SCREAMING_SNAKE_CASE						=		TFResNetBottleNeckLayer if config.layer_type == '''bottleneck''' else TFResNetBasicLayer

           SCREAMING_SNAKE_CASE						=		[layer(_lowerCamelCase		,_lowerCamelCase		,stride=_lowerCamelCase		,activation=config.hidden_act		,name="""layers.0"""			)]
           layers += [
               layer(_lowerCamelCase		,_lowerCamelCase		,activation=config.hidden_act		,name=F"""layers.{i + 1}"""			)
               for i in range(depth - 1			)
           ]
           SCREAMING_SNAKE_CASE						=		layers






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Optional[int]		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       bool = False			)	->						Dict:




           '''simple docstring'''


           for layer in self.stage_layers:
                 SCREAMING_SNAKE_CASE						=		layer(_lowerCamelCase		,training=_lowerCamelCase			)
           return hidden_state

class 						UpperCamelCase__					( tf.keras.layers.Layer  ):



     '''simple docstring'''

     def __init__(				self		:       Any		,lowerCamelCase__		:       ResNetConfig		,**lowerCamelCase__		:       List[Any]			)	->						Any:




           '''simple docstring'''


           super().__init__(**_lowerCamelCase			)
           # based on `downsample_in_first_stage` the first layer of the first stage may or may not downsample the input
           SCREAMING_SNAKE_CASE						=		[
               TFResNetStage(
                   _lowerCamelCase		,config.embedding_size		,config.hidden_sizes[0]		,stride=2 if config.downsample_in_first_stage else 1		,depth=config.depths[0]		,name="""stages.0"""		,)
           ]
           for i, (in_channels, out_channels, depth) in enumerate(
               zip(config.hidden_sizes		,config.hidden_sizes[1:]		,config.depths[1:]			)			):
                 self.stages.append(TFResNetStage(_lowerCamelCase		,_lowerCamelCase		,_lowerCamelCase		,depth=_lowerCamelCase		,name=F"""stages.{i + 1}"""			)			)






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Optional[int]		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       bool = False		,lowerCamelCase__		:       bool = True		,lowerCamelCase__		:       bool = False		,)	->						int:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		() if output_hidden_states else None

           for stage_module in self.stages:
                 if output_hidden_states:
                       SCREAMING_SNAKE_CASE						=		hidden_states + (hidden_state,)

                 SCREAMING_SNAKE_CASE						=		stage_module(_lowerCamelCase		,training=_lowerCamelCase			)

           if output_hidden_states:
                 SCREAMING_SNAKE_CASE						=		hidden_states + (hidden_state,)

           if not return_dict:
                 return tuple(v for v in [hidden_state, hidden_states] if v is not None			)

           return TFBaseModelOutputWithNoAttention(last_hidden_state=_lowerCamelCase		,hidden_states=_lowerCamelCase			)

class 						UpperCamelCase__					( __UpperCAmelCase  ):



     '''simple docstring'''

     __snake_case       :      Tuple						      =     ResNetConfig
     __snake_case       :      str						      =     """resnet"""
     __snake_case       :      str						      =     """pixel_values"""






     @property
     def 							SCREAMING_SNAKE_CASE__    (				self		:       List[str]			)	->						Union[str, Any]:




           '''simple docstring'''


           return {"pixel_values": tf.TensorSpec(shape=(None, self.config.num_channels, 224, 224)		,dtype=tf.floataa			)}
SCREAMING_SNAKE_CASE_		  =			r'\n    This model is a TensorFlow\n    [tf.keras.layers.Layer](https://www.tensorflow.org/api_docs/python/tf/keras/layers/Layer) sub-class. Use it as a\n    regular TensorFlow Module and refer to the TensorFlow documentation for all matter related to general usage and\n    behavior.\n\n    Parameters:\n        config ([`ResNetConfig`]): Model configuration class with all the parameters of the model.\n            Initializing with a config file does not load the weights associated with the model, only the\n            configuration. Check out the [`~TFPreTrainedModel.from_pretrained`] method to load the model weights.\n'


SCREAMING_SNAKE_CASE_		  =			r'\n    Args:\n        pixel_values (`tf.Tensor` of shape `(batch_size, num_channels, height, width)`):\n            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See\n            [`ConvNextImageProcessor.__call__`] for details.\n\n        output_hidden_states (`bool`, *optional*):\n            Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for\n            more detail.\n        return_dict (`bool`, *optional*):\n            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.\n'

@keras_serializable
class 						UpperCamelCase__					( tf.keras.layers.Layer  ):



     '''simple docstring'''

     __snake_case       :      Optional[Any]						      =     ResNetConfig






     def __init__(				self		:       int		,lowerCamelCase__		:       ResNetConfig		,**lowerCamelCase__		:       Tuple			)	->						Dict:




           '''simple docstring'''


           super().__init__(**_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		config
           SCREAMING_SNAKE_CASE						=		TFResNetEmbeddings(_lowerCamelCase		,name="""embedder"""			)
           SCREAMING_SNAKE_CASE						=		TFResNetEncoder(_lowerCamelCase		,name="""encoder"""			)
           SCREAMING_SNAKE_CASE						=		tf.keras.layers.GlobalAveragePoolingaD(keepdims=_lowerCamelCase			)






     @unpack_inputs
     def 							SCREAMING_SNAKE_CASE__    (				self		:       Optional[Any]		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       Optional[bool] = None		,lowerCamelCase__		:       Optional[bool] = None		,lowerCamelCase__		:       bool = False		,)	->						Optional[int]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		(
               output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
           )
           SCREAMING_SNAKE_CASE						=		return_dict if return_dict is not None else self.config.use_return_dict

           # TF 2.0 image layers can't use NCHW format when running on CPU.
           # We transpose to NHWC format and then transpose back after the full forward pass.
           # (batch_size, num_channels, height, width) -> (batch_size, height, width, num_channels)
           SCREAMING_SNAKE_CASE						=		tf.transpose(_lowerCamelCase		,perm=[0, 2, 3, 1]			)
           SCREAMING_SNAKE_CASE						=		self.embedder(_lowerCamelCase		,training=_lowerCamelCase			)

           SCREAMING_SNAKE_CASE						=		self.encoder(
               _lowerCamelCase		,output_hidden_states=_lowerCamelCase		,return_dict=_lowerCamelCase		,training=_lowerCamelCase			)

           SCREAMING_SNAKE_CASE						=		encoder_outputs[0]

           SCREAMING_SNAKE_CASE						=		self.pooler(_lowerCamelCase			)

           # Transpose all the outputs to the NCHW format
           # (batch_size, height, width, num_channels) -> (batch_size, num_channels, height, width)
           SCREAMING_SNAKE_CASE						=		tf.transpose(_lowerCamelCase		,(0, 3, 1, 2)			)
           SCREAMING_SNAKE_CASE						=		tf.transpose(_lowerCamelCase		,(0, 3, 1, 2)			)
           SCREAMING_SNAKE_CASE						=		()
           for hidden_state in encoder_outputs[1:]:
                 SCREAMING_SNAKE_CASE						=		hidden_states + tuple(tf.transpose(_lowerCamelCase		,(0, 3, 1, 2)			) for h in hidden_state			)

           if not return_dict:
                 return (last_hidden_state, pooled_output) + hidden_states

           SCREAMING_SNAKE_CASE						=		hidden_states if output_hidden_states else None

           return TFBaseModelOutputWithPoolingAndNoAttention(
               last_hidden_state=_lowerCamelCase		,pooler_output=_lowerCamelCase		,hidden_states=_lowerCamelCase		,)

@add_start_docstrings(
    "The bare ResNet model outputting raw features without any specific head on top."				, __UpperCAmelCase				, )
class 						UpperCamelCase__					( __UpperCAmelCase  ):



     '''simple docstring'''

     def __init__(				self		:       str		,lowerCamelCase__		:       ResNetConfig		,**lowerCamelCase__		:       Any			)	->						Any:




           '''simple docstring'''


           super().__init__(_lowerCamelCase		,**_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		TFResNetMainLayer(config=_lowerCamelCase		,name="""resnet"""			)






     @add_start_docstrings_to_model_forward(_lowerCamelCase			)
     @add_code_sample_docstrings(
         checkpoint=_CHECKPOINT_FOR_DOC		,output_type=_lowerCamelCase		,config_class=_CONFIG_FOR_DOC		,modality="""vision"""		,expected_output=_EXPECTED_OUTPUT_SHAPE		,)
     @unpack_inputs
     def 							SCREAMING_SNAKE_CASE__    (				self		:       str		,lowerCamelCase__		:       tf.Tensor		,lowerCamelCase__		:       Optional[bool] = None		,lowerCamelCase__		:       Optional[bool] = None		,lowerCamelCase__		:       bool = False		,)	->						Any:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		(
               output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
           )
           SCREAMING_SNAKE_CASE						=		return_dict if return_dict is not None else self.config.use_return_dict

           SCREAMING_SNAKE_CASE						=		self.resnet(
               pixel_values=_lowerCamelCase		,output_hidden_states=_lowerCamelCase		,return_dict=_lowerCamelCase		,training=_lowerCamelCase		,)
           return resnet_outputs

@add_start_docstrings(
    "\n    ResNet Model with an image classification head on top (a linear layer on top of the pooled features), e.g. for\n    ImageNet.\n    "				, __UpperCAmelCase				, )
class 						UpperCamelCase__					( __UpperCAmelCase				, __UpperCAmelCase  ):



     '''simple docstring'''

     def __init__(				self		:       Union[str, Any]		,lowerCamelCase__		:       ResNetConfig		,**lowerCamelCase__		:       Any			)	->						Dict:




           '''simple docstring'''


           super().__init__(_lowerCamelCase		,**_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		config.num_labels
           SCREAMING_SNAKE_CASE						=		TFResNetMainLayer(_lowerCamelCase		,name="""resnet"""			)
           # classification head
           SCREAMING_SNAKE_CASE						=		(
               tf.keras.layers.Dense(config.num_labels		,name="""classifier.1"""			)
               if config.num_labels > 0
               else tf.keras.layers.Activation("""linear"""		,name="""classifier.1"""			)
           )






     def 							SCREAMING_SNAKE_CASE__    (				self		:       Union[str, Any]		,lowerCamelCase__		:       tf.Tensor			)	->						str:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		tf.keras.layers.Flatten()(_lowerCamelCase			)
           SCREAMING_SNAKE_CASE						=		self.classifier_layer(_lowerCamelCase			)
           return logits






     @add_start_docstrings_to_model_forward(_lowerCamelCase			)
     @add_code_sample_docstrings(
         checkpoint=_IMAGE_CLASS_CHECKPOINT		,output_type=_lowerCamelCase		,config_class=_CONFIG_FOR_DOC		,expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT		,)
     @unpack_inputs
     def 							SCREAMING_SNAKE_CASE__    (				self		:       Any		,lowerCamelCase__		:       tf.Tensor = None		,lowerCamelCase__		:       tf.Tensor = None		,lowerCamelCase__		:       bool = None		,lowerCamelCase__		:       bool = None		,lowerCamelCase__		:       bool = False		,)	->						List[Any]:




           '''simple docstring'''


           SCREAMING_SNAKE_CASE						=		return_dict if return_dict is not None else self.config.use_return_dict

           SCREAMING_SNAKE_CASE						=		self.resnet(
               _lowerCamelCase		,output_hidden_states=_lowerCamelCase		,return_dict=_lowerCamelCase		,training=_lowerCamelCase			)

           SCREAMING_SNAKE_CASE						=		outputs.pooler_output if return_dict else outputs[1]

           SCREAMING_SNAKE_CASE						=		self.classifier(_lowerCamelCase			)

           SCREAMING_SNAKE_CASE						=		None if labels is None else self.hf_compute_loss(_lowerCamelCase		,_lowerCamelCase			)

           if not return_dict:
                 SCREAMING_SNAKE_CASE						=		(logits,) + outputs[2:]
                 return (loss,) + output if loss is not None else output

           return TFImageClassifierOutputWithNoAttention(loss=_lowerCamelCase		,logits=_lowerCamelCase		,hidden_states=outputs.hidden_states			)








"""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 		lowercase_   (		_UpperCAmelCase						):




	"""simple docstring"""






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

def 		lowercase_   (		_UpperCAmelCase						):




	"""simple docstring"""






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

def 		lowercase_   (		_UpperCAmelCase						):




	"""simple docstring"""






	A_	:					Optional[Any]						 =		create_text_encoder_lora_attn_procs(_UpperCAmelCase						)
	A_	:					List[str]						 =		AttnProcsLayers(_UpperCAmelCase						)
	return text_encoder_lora_layers

def 		lowercase_   (		_UpperCAmelCase	,   _UpperCAmelCase=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'''						):
				A_	:					List[Any]						 =		(
				    torch.randn_like(layer_module.up.weight						)
				    if randn_weight
				    else torch.zeros_like(layer_module.up.weight						)
				)
				A_	:					Tuple						 =		torch.nn.Parameter(_UpperCAmelCase						)

class 							lowercase   (     unittest.TestCase):







				def        a_	(				self :       List[Any]							):



					"""simple docstring"""

					torch.manual_seed(0							)
					A_	:					List[str]						 =		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   ,  )
					A_	:					Optional[int]						 =		DDIMScheduler(
					    beta_start=0.00085   ,  beta_end=0.012   ,  beta_schedule='''scaled_linear'''   ,  clip_sample=_lowerCamelCase   ,  set_alpha_to_one=_lowerCamelCase   ,  steps_offset=1   ,  )
					torch.manual_seed(0							)
					A_	:					Any						 =		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   ,  )
					A_	:					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=10_00   ,  )
					A_	:					List[Any]						 =		CLIPTextModel(_lowerCamelCase							)
					A_	:					str						 =		CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''							)

					A_ ,			A_	:					Any						 =		create_unet_lora_layers(_lowerCamelCase							)
					A_	:					Tuple						 =		create_text_encoder_lora_layers(_lowerCamelCase							)

					A_	:					Any						 =		{
					    '''unet''': unet,
					    '''scheduler''': scheduler,
					    '''vae''': vae,
					    '''text_encoder''': text_encoder,
					    '''tokenizer''': tokenizer,
					    '''safety_checker''': None,
					    '''feature_extractor''': None,
					}
					A_	:					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        a_	(				self :       str							):



					"""simple docstring"""

					A_	:					Optional[int]						 =		1
					A_	:					Tuple						 =		10
					A_	:					int						 =		4
					A_	:					List[Any]						 =		(32, 32)

					A_	:					str						 =		torch.manual_seed(0							)
					A_	:					Dict						 =		floats_tensor((batch_size, num_channels) + sizes							)
					A_	:					List[str]						 =		torch.randint(1   ,  _lowerCamelCase   ,  size=(batch_size, sequence_length)   ,  generator=_lowerCamelCase							)

					A_	:					int						 =		{
					    '''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        a_	(				self :       str							):



					"""simple docstring"""

					A_	:					Union[str, Any]						 =		77

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

					A_	:					Optional[Any]						 =		{}
					A_	:					int						 =		inputs
					return prepared_inputs







				def        a_	(				self :       Optional[int]   ,  _lowerCamelCase :       List[Any]							):



					"""simple docstring"""

					A_ ,			A_	:					List[str]						 =		self.get_dummy_components()
					LoraLoaderMixin.save_lora_weights(
					    save_directory=_lowerCamelCase   ,  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(_lowerCamelCase   ,  '''pytorch_lora_weights.bin'''							)							)							)







				def        a_	(				self :       Any							):



					"""simple docstring"""

					A_ ,			A_	:					Optional[Any]						 =		self.get_dummy_components()
					A_	:					List[str]						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					List[str]						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

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

					A_	:					Optional[int]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					str						 =		original_images[0, -3:, -3:, -1]

					with tempfile.TemporaryDirectory() as tmpdirname:
						LoraLoaderMixin.save_lora_weights(
						    save_directory=_lowerCamelCase   ,  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(_lowerCamelCase   ,  '''pytorch_lora_weights.bin'''							)							)							)
						sd_pipe.load_lora_weights(_lowerCamelCase							)

					A_	:					List[Any]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					str						 =		lora_images[0, -3:, -3:, -1]

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







				def        a_	(				self :       Optional[int]							):



					"""simple docstring"""

					A_ ,			A_	:					List[Any]						 =		self.get_dummy_components()
					A_	:					str						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					Any						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

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

					A_	:					Union[str, Any]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					int						 =		original_images[0, -3:, -3:, -1]

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

					A_	:					Union[str, Any]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Dict						 =		lora_images[0, -3:, -3:, -1]

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







				def        a_	(				self :       Optional[int]							):



					"""simple docstring"""

					A_ ,			A_	:					List[Any]						 =		self.get_dummy_components()
					A_	:					Any						 =		lora_components['''unet_lora_attn_procs''']
					A_	:					Any						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					Tuple						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

					A_ ,			A_ ,			A_	:					Tuple						 =		self.get_dummy_inputs()

					A_	:					Tuple						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Optional[Any]						 =		original_images[0, -3:, -3:, -1]

					with tempfile.TemporaryDirectory() as tmpdirname:
						A_	:					Tuple						 =		sd_pipe.unet
						unet.set_attn_processor(_lowerCamelCase							)
						unet.save_attn_procs(_lowerCamelCase							)
						self.assertTrue(os.path.isfile(os.path.join(_lowerCamelCase   ,  '''pytorch_lora_weights.bin'''							)							)							)
						sd_pipe.load_lora_weights(_lowerCamelCase							)

					A_	:					Any						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Optional[int]						 =		lora_images[0, -3:, -3:, -1]

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







				def        a_	(				self :       Tuple							):



					"""simple docstring"""

					A_ ,			A_	:					List[str]						 =		self.get_dummy_components()
					A_	:					Dict						 =		StableDiffusionPipeline(**_lowerCamelCase							)

					A_	:					Any						 =		self.get_dummy_tokens()

					# inference without lora
					A_	:					str						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_without_lora.shape == (1, 77, 32)

					# create lora_attn_procs with zeroed out up.weights
					A_	:					str						 =		create_text_encoder_lora_attn_procs(pipe.text_encoder							)
					set_lora_up_weights(_lowerCamelCase   ,  randn_weight=_lowerCamelCase							)

					# monkey patch
					pipe._modify_text_encoder(_lowerCamelCase							)

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

					# inference with lora
					A_	:					List[str]						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_with_lora.shape == (1, 77, 32)

					assert torch.allclose(
					    _lowerCamelCase   ,  _lowerCamelCase							), "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
					A_	:					Tuple						 =		create_text_encoder_lora_attn_procs(pipe.text_encoder							)
					set_lora_up_weights(_lowerCamelCase   ,  randn_weight=_lowerCamelCase							)

					# monkey patch
					pipe._modify_text_encoder(_lowerCamelCase							)

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

					# inference with lora
					A_	:					str						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_with_lora.shape == (1, 77, 32)

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







				def        a_	(				self :       Optional[int]							):



					"""simple docstring"""

					A_ ,			A_	:					List[Any]						 =		self.get_dummy_components()
					A_	:					Optional[int]						 =		StableDiffusionPipeline(**_lowerCamelCase							)

					A_	:					List[Any]						 =		self.get_dummy_tokens()

					# inference without lora
					A_	:					Tuple						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_without_lora.shape == (1, 77, 32)

					# create lora_attn_procs with randn up.weights
					A_	:					List[Any]						 =		create_text_encoder_lora_attn_procs(pipe.text_encoder							)
					set_lora_up_weights(_lowerCamelCase   ,  randn_weight=_lowerCamelCase							)

					# monkey patch
					pipe._modify_text_encoder(_lowerCamelCase							)

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

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

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

					# remove monkey patch
					pipe._remove_text_encoder_monkey_patch()

					# inference with removed lora
					A_	:					Optional[int]						 =		pipe.text_encoder(**_lowerCamelCase							)[0]
					assert outputs_without_lora_removed.shape == (1, 77, 32)

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







				def        a_	(				self :       Tuple							):



					"""simple docstring"""

					A_ ,			A_	:					Dict						 =		self.get_dummy_components()
					A_	:					str						 =		StableDiffusionPipeline(**_lowerCamelCase							)
					A_	:					Dict						 =		sd_pipe.to(_lowerCamelCase							)
					sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

					A_ ,			A_ ,			A_	:					Tuple						 =		self.get_dummy_inputs()

					with tempfile.TemporaryDirectory() as tmpdirname:
						LoraLoaderMixin.save_lora_weights(
						    save_directory=_lowerCamelCase   ,  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(_lowerCamelCase   ,  '''pytorch_lora_weights.bin'''							)							)							)
						sd_pipe.load_lora_weights(_lowerCamelCase							)

					A_	:					Dict						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Union[str, Any]						 =		lora_images[0, -3:, -3:, -1]

					A_	:					int						 =		sd_pipe(**_lowerCamelCase   ,  cross_attention_kwargs={'''scale''': 0.5}							).images
					A_	:					int						 =		lora_images_with_scale[0, -3:, -3:, -1]

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







				def        a_	(				self :       Tuple							):



					"""simple docstring"""

					with tempfile.TemporaryDirectory() as tmpdirname:
						self.create_lora_weight_file(_lowerCamelCase							)

						A_ ,			A_	:					str						 =		self.get_dummy_components()
						A_	:					Optional[Any]						 =		StableDiffusionPipeline(**_lowerCamelCase							)
						A_	:					List[Any]						 =		sd_pipe.to(_lowerCamelCase							)
						sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

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

            # load LoRA weight file
						sd_pipe.load_lora_weights(_lowerCamelCase							)

						# check if lora attention processors are used
						for _, module in sd_pipe.unet.named_modules():
							if isinstance(_lowerCamelCase   ,  _lowerCamelCase							):
								A_	:					Any						 =		(
								    LoRAAttnProcessor2_0 if hasattr(_lowerCamelCase   ,  '''scaled_dot_product_attention'''							) else LoRAAttnProcessor
								)
								self.assertIsInstance(module.processor   ,  _lowerCamelCase							)







				@unittest.skipIf(torch_device != '''cuda'''   ,  '''This test is supposed to run on GPU'''							)
				def        a_	(				self :       Optional[Any]							):



					"""simple docstring"""

					with tempfile.TemporaryDirectory() as tmpdirname:
						self.create_lora_weight_file(_lowerCamelCase							)

						A_ ,			A_	:					Tuple						 =		self.get_dummy_components()
						A_	:					Optional[int]						 =		StableDiffusionPipeline(**_lowerCamelCase							)
						A_	:					Any						 =		sd_pipe.to(_lowerCamelCase							)
						sd_pipe.set_progress_bar_config(disable=_lowerCamelCase							)

						# 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(_lowerCamelCase   ,  _lowerCamelCase							):
								self.assertIsInstance(module.processor   ,  _lowerCamelCase							)

            # load LoRA weight file
						sd_pipe.load_lora_weights(_lowerCamelCase							)

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







				@unittest.skipIf(torch_device != '''cuda'''   ,  '''This test is supposed to run on GPU'''							)
				def        a_	(				self :       Optional[Any]							):



					"""simple docstring"""

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

					A_ ,			A_ ,			A_	:					int						 =		self.get_dummy_inputs()

					# enable XFormers
					sd_pipe.enable_xformers_memory_efficient_attention()

					A_	:					str						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					Tuple						 =		original_images[0, -3:, -3:, -1]

					with tempfile.TemporaryDirectory() as tmpdirname:
						LoraLoaderMixin.save_lora_weights(
						    save_directory=_lowerCamelCase   ,  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(_lowerCamelCase   ,  '''pytorch_lora_weights.bin'''							)							)							)
						sd_pipe.load_lora_weights(_lowerCamelCase							)

					A_	:					List[Any]						 =		sd_pipe(**_lowerCamelCase							).images
					A_	:					int						 =		lora_images[0, -3:, -3:, -1]

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





