

from typing import Optional

import torch
import torch.utils.checkpoint
from torch import Tensor, nn
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss

from ...activations import ACTaFN
from ...modeling_outputs import (
    BackboneOutput,
    BaseModelOutputWithNoAttention,
    BaseModelOutputWithPoolingAndNoAttention,
    ImageClassifierOutputWithNoAttention,
)
from ...modeling_utils import PreTrainedModel
from ...utils import (
    add_code_sample_docstrings,
    add_start_docstrings,
    add_start_docstrings_to_model_forward,
    logging,
    replace_return_docstrings,
)
from ...utils.backbone_utils import BackboneMixin
from .configuration_resnet import ResNetConfig


_UpperCAmelCase		:       Any						    =	logging.get_logger(__name__)

# General docstring
_UpperCAmelCase		:       Optional[Any]						    =	'''ResNetConfig'''

# Base docstring
_UpperCAmelCase		:       Optional[Any]						    =	'''microsoft/resnet-50'''
_UpperCAmelCase		:       Dict						    =	[1, 2048, 7, 7]

# Image classification docstring
_UpperCAmelCase		:       Optional[Any]						    =	'''microsoft/resnet-50'''
_UpperCAmelCase		:       Dict						    =	'''tiger cat'''

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







class 			lowercase    (	nn.Module      ):




    def __init__( self				,       snake_case				,       snake_case				,       snake_case = 3				,       snake_case = 1				,       snake_case = "relu"	):
           super().__init__()
           snake_case_          =   nn.Convad(
               lowerCamelCase__				,       lowerCamelCase__				,       kernel_size=lowerCamelCase__				,       stride=lowerCamelCase__				,       padding=kernel_size // 2				,       bias=lowerCamelCase__	)
           snake_case_          =   nn.BatchNormad(lowerCamelCase__	)
           snake_case_          =   ACTaFN[activation] if activation is not None else nn.Identity()




    def    a							( self				,       snake_case	):
           snake_case_          =   self.convolution(lowerCamelCase__	)
           snake_case_          =   self.normalization(lowerCamelCase__	)
           snake_case_          =   self.activation(lowerCamelCase__	)
           return hidden_state







class 			lowercase    (	nn.Module      ):




    def __init__( self				,       snake_case	):
           super().__init__()
           snake_case_          =   ResNetConvLayer(
               config.num_channels				,       config.embedding_size				,       kernel_size=7				,       stride=2				,       activation=config.hidden_act	)
           snake_case_          =   nn.MaxPoolad(kernel_size=3				,       stride=2				,       padding=1	)
           snake_case_          =   config.num_channels




    def    a							( self				,       snake_case	):
           snake_case_          =   pixel_values.shape[1]
           if 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.'	)
           snake_case_          =   self.embedder(lowerCamelCase__	)
           snake_case_          =   self.pooler(lowerCamelCase__	)
           return embedding







class 			lowercase    (	nn.Module      ):




    def __init__( self				,       snake_case				,       snake_case				,       snake_case = 2	):
           super().__init__()
           snake_case_          =   nn.Convad(lowerCamelCase__				,       lowerCamelCase__				,       kernel_size=1				,       stride=lowerCamelCase__				,       bias=lowerCamelCase__	)
           snake_case_          =   nn.BatchNormad(lowerCamelCase__	)




    def    a							( self				,       snake_case	):
           snake_case_          =   self.convolution(lowerCamelCase__	)
           snake_case_          =   self.normalization(lowerCamelCase__	)
           return hidden_state







class 			lowercase    (	nn.Module      ):




    def __init__( self				,       snake_case				,       snake_case				,       snake_case = 1				,       snake_case = "relu"	):
           super().__init__()
           snake_case_          =   in_channels != out_channels or stride != 1
           snake_case_          =   (
               ResNetShortCut(lowerCamelCase__				,       lowerCamelCase__				,       stride=lowerCamelCase__	) if should_apply_shortcut else nn.Identity()
           )
           snake_case_          =   nn.Sequential(
               ResNetConvLayer(lowerCamelCase__				,       lowerCamelCase__				,       stride=lowerCamelCase__	)				,       ResNetConvLayer(lowerCamelCase__				,       lowerCamelCase__				,       activation=lowerCamelCase__	)				,       )
           snake_case_          =   ACTaFN[activation]




    def    a							( self				,       snake_case	):
           snake_case_          =   hidden_state
           snake_case_          =   self.layer(lowerCamelCase__	)
           snake_case_          =   self.shortcut(lowerCamelCase__	)
           hidden_state += residual
           snake_case_          =   self.activation(lowerCamelCase__	)
           return hidden_state







class 			lowercase    (	nn.Module      ):




    def __init__( self				,       snake_case				,       snake_case				,       snake_case = 1				,       snake_case = "relu"				,       snake_case = 4	):
           super().__init__()
           snake_case_          =   in_channels != out_channels or stride != 1
           snake_case_          =   out_channels // reduction
           snake_case_          =   (
               ResNetShortCut(lowerCamelCase__				,       lowerCamelCase__				,       stride=lowerCamelCase__	) if should_apply_shortcut else nn.Identity()
           )
           snake_case_          =   nn.Sequential(
               ResNetConvLayer(lowerCamelCase__				,       lowerCamelCase__				,       kernel_size=1	)				,       ResNetConvLayer(lowerCamelCase__				,       lowerCamelCase__				,       stride=lowerCamelCase__	)				,       ResNetConvLayer(lowerCamelCase__				,       lowerCamelCase__				,       kernel_size=1				,       activation=lowerCamelCase__	)				,       )
           snake_case_          =   ACTaFN[activation]




    def    a							( self				,       snake_case	):
           snake_case_          =   hidden_state
           snake_case_          =   self.layer(lowerCamelCase__	)
           snake_case_          =   self.shortcut(lowerCamelCase__	)
           hidden_state += residual
           snake_case_          =   self.activation(lowerCamelCase__	)
           return hidden_state







class 			lowercase    (	nn.Module      ):




    def __init__( self				,       snake_case				,       snake_case				,       snake_case				,       snake_case = 2				,       snake_case = 2				,       ):
           super().__init__()

           snake_case_          =   ResNetBottleNeckLayer if config.layer_type == 'bottleneck' else ResNetBasicLayer

           snake_case_          =   nn.Sequential(
               # downsampling is done in the first layer with stride of 2
               layer(lowerCamelCase__				,       lowerCamelCase__				,       stride=lowerCamelCase__				,       activation=config.hidden_act	)				,       *[layer(lowerCamelCase__				,       lowerCamelCase__				,       activation=config.hidden_act	) for _ in range(depth - 1	)]				,       )




    def    a							( self				,       snake_case	):
           snake_case_          =   input
           for layer in self.layers:
                  snake_case_          =   layer(lowerCamelCase__	)
           return hidden_state







class 			lowercase    (	nn.Module      ):




    def __init__( self				,       snake_case	):
           super().__init__()
           snake_case_          =   nn.ModuleList([]	)
           # based on `downsample_in_first_stage` the first layer of the first stage may or may not downsample the input
           self.stages.append(
               ResNetStage(
                   lowerCamelCase__				,       config.embedding_size				,       config.hidden_sizes[0]				,       stride=2 if config.downsample_in_first_stage else 1				,       depth=config.depths[0]				,       )	)
           snake_case_          =   zip(config.hidden_sizes				,       config.hidden_sizes[1:]	)
           for (in_channels, out_channels), depth in zip(lowerCamelCase__				,       config.depths[1:]	):
                  self.stages.append(ResNetStage(lowerCamelCase__				,       lowerCamelCase__				,       lowerCamelCase__				,       depth=lowerCamelCase__	)	)




    def    a							( self				,       snake_case				,       snake_case = False				,       snake_case = True	):
           snake_case_          =   () if output_hidden_states else None

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

                  snake_case_          =   stage_module(lowerCamelCase__	)

           if output_hidden_states:
                  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 BaseModelOutputWithNoAttention(
               last_hidden_state=lowerCamelCase__				,       hidden_states=lowerCamelCase__				,       )







class 			lowercase    (	A__      ):
    __SCREAMING_SNAKE_CASE :			Optional[Any]	      =      ResNetConfig
    __SCREAMING_SNAKE_CASE :			Tuple	      =      'resnet'
    __SCREAMING_SNAKE_CASE :			int	      =      'pixel_values'
    __SCREAMING_SNAKE_CASE :			List[Any]	      =      True




    def    a							( self				,       snake_case	):
           if isinstance(lowerCamelCase__				,       nn.Convad	):
                  nn.init.kaiming_normal_(module.weight				,       mode='fan_out'				,       nonlinearity='relu'	)
           elif isinstance(lowerCamelCase__				,       (nn.BatchNormad, nn.GroupNorm)	):
                  nn.init.constant_(module.weight				,       1	)
                  nn.init.constant_(module.bias				,       0	)




    def    a							( self				,       snake_case				,       snake_case=False	):
           if isinstance(lowerCamelCase__				,       lowerCamelCase__	):
                  snake_case_          =   value



_UpperCAmelCase		:       Optional[Any]						    =	R'''
    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass. Use it
    as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and
    behavior.

    Parameters:
        config ([`ResNetConfig`]): Model configuration class with all the parameters of the model.
            Initializing with a config file does not load the weights associated with the model, only the
            configuration. Check out the [`~PreTrainedModel.from_pretrained`] method to load the model weights.
'''

_UpperCAmelCase		:       Optional[Any]						    =	R'''
    Args:
        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):
            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See
            [`ConvNextImageProcessor.__call__`] for details.

        output_hidden_states (`bool`, *optional*):
            Whether or not to return the hidden states of all layers. See `hidden_states` under returned tensors for
            more detail.
        return_dict (`bool`, *optional*):
            Whether or not to return a [`~utils.ModelOutput`] instead of a plain tuple.
'''







@add_start_docstrings(
    '''The bare ResNet model outputting raw features without any specific head on top.'''     ,  A__     ,  )
class 			lowercase    (	A__      ):




    def __init__( self				,       snake_case	):
           super().__init__(lowerCamelCase__	)
           snake_case_          =   config
           snake_case_          =   ResNetEmbeddings(lowerCamelCase__	)
           snake_case_          =   ResNetEncoder(lowerCamelCase__	)
           snake_case_          =   nn.AdaptiveAvgPoolad((1, 1)	)
           # Initialize weights and apply final processing
           self.post_init()




    @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				,       )
    def    a							( self				,       snake_case				,       snake_case = None				,       snake_case = None	):
           snake_case_          =   (
               output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
           )
           snake_case_          =   return_dict if return_dict is not None else self.config.use_return_dict

           snake_case_          =   self.embedder(lowerCamelCase__	)

           snake_case_          =   self.encoder(
               lowerCamelCase__				,       output_hidden_states=lowerCamelCase__				,       return_dict=lowerCamelCase__	)

           snake_case_          =   encoder_outputs[0]

           snake_case_          =   self.pooler(lowerCamelCase__	)

           if not return_dict:
                  return (last_hidden_state, pooled_output) + encoder_outputs[1:]

           return BaseModelOutputWithPoolingAndNoAttention(
               last_hidden_state=lowerCamelCase__				,       pooler_output=lowerCamelCase__				,       hidden_states=encoder_outputs.hidden_states				,       )







@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    '''     ,  A__     ,  )
class 			lowercase    (	A__      ):




    def __init__( self				,       snake_case	):
           super().__init__(lowerCamelCase__	)
           snake_case_          =   config.num_labels
           snake_case_          =   ResNetModel(lowerCamelCase__	)
           # classification head
           snake_case_          =   nn.Sequential(
               nn.Flatten()				,       nn.Linear(config.hidden_sizes[-1]				,       config.num_labels	) if config.num_labels > 0 else nn.Identity()				,       )
           # initialize weights and apply final processing
           self.post_init()




    @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				,       )
    def    a							( self				,       snake_case = None				,       snake_case = None				,       snake_case = None				,       snake_case = None				,       ):
           snake_case_          =   return_dict if return_dict is not None else self.config.use_return_dict

           snake_case_          =   self.resnet(lowerCamelCase__				,       output_hidden_states=lowerCamelCase__				,       return_dict=lowerCamelCase__	)

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

           snake_case_          =   self.classifier(lowerCamelCase__	)

           snake_case_          =   None

           if labels is not None:
                  if self.config.problem_type is None:
                         if self.num_labels == 1:
                                snake_case_          =   'regression'
                         elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int):
                                snake_case_          =   'single_label_classification'
                         else:
                                snake_case_          =   'multi_label_classification'
                  if self.config.problem_type == "regression":
                         snake_case_          =   MSELoss()
                         if self.num_labels == 1:
                                snake_case_          =   loss_fct(logits.squeeze()				,       labels.squeeze()	)
                         else:
                                snake_case_          =   loss_fct(lowerCamelCase__				,       lowerCamelCase__	)
                  elif self.config.problem_type == "single_label_classification":
                         snake_case_          =   CrossEntropyLoss()
                         snake_case_          =   loss_fct(logits.view(-1				,       self.num_labels	)				,       labels.view(-1	)	)
                  elif self.config.problem_type == "multi_label_classification":
                         snake_case_          =   BCEWithLogitsLoss()
                         snake_case_          =   loss_fct(lowerCamelCase__				,       lowerCamelCase__	)

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

           return ImageClassifierOutputWithNoAttention(loss=lowerCamelCase__				,       logits=lowerCamelCase__				,       hidden_states=outputs.hidden_states	)







@add_start_docstrings(
    '''\n    ResNet backbone, to be used with frameworks like DETR and MaskFormer.\n    '''     ,  A__     ,  )
class 			lowercase    (	A__     ,  A__      ):




    def __init__( self				,       snake_case	):
           super().__init__(lowerCamelCase__	)
           super()._init_backbone(lowerCamelCase__	)

           snake_case_          =   [config.embedding_size] + config.hidden_sizes
           snake_case_          =   ResNetEmbeddings(lowerCamelCase__	)
           snake_case_          =   ResNetEncoder(lowerCamelCase__	)

           # initialize weights and apply final processing
           self.post_init()




    @add_start_docstrings_to_model_forward(lowerCamelCase__	)
    @replace_return_docstrings(output_type=lowerCamelCase__				,       config_class=_CONFIG_FOR_DOC	)
    def    a							( self				,       snake_case				,       snake_case = None				,       snake_case = None	):
           snake_case_          =   return_dict if return_dict is not None else self.config.use_return_dict
           snake_case_          =   (
               output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
           )

           snake_case_          =   self.embedder(lowerCamelCase__	)

           snake_case_          =   self.encoder(lowerCamelCase__				,       output_hidden_states=lowerCamelCase__				,       return_dict=lowerCamelCase__	)

           snake_case_          =   outputs.hidden_states

           snake_case_          =   ()
           for idx, stage in enumerate(self.stage_names	):
                  if stage in self.out_features:
                         feature_maps += (hidden_states[idx],)

           if not return_dict:
                  snake_case_          =   (feature_maps,)
                  if output_hidden_states:
                         output += (outputs.hidden_states,)
                  return output

           return BackboneOutput(
               feature_maps=lowerCamelCase__				,       hidden_states=outputs.hidden_states if output_hidden_states else None				,       attentions=lowerCamelCase__				,       )







import gc
import unittest

from diffusers import FlaxControlNetModel, FlaxStableDiffusionControlNetPipeline
from diffusers.utils import is_flax_available, load_image, slow
from diffusers.utils.testing_utils import require_flax


if is_flax_available():
					import jax
					import jax.numpy as jnp
					from flax.jax_utils import replicate
					from flax.training.common_utils import shard







@slow
@require_flax
class 			lowercase    (	unittest.TestCase      ):




				def    a							( self	):
											# clean up the VRAM after each test
											super().tearDown()
											gc.collect()




				def    a							( self	):
											snake_case_    ,   snake_case_          =   FlaxControlNetModel.from_pretrained(
											    'lllyasviel/sd-controlnet-canny'				,       from_pt=snake_case				,       dtype=jnp.bfloataa	)
											snake_case_    ,   snake_case_          =   FlaxStableDiffusionControlNetPipeline.from_pretrained(
											    'runwayml/stable-diffusion-v1-5'				,       controlnet=snake_case				,       from_pt=snake_case				,       dtype=jnp.bfloataa	)
											snake_case_          =   controlnet_params

											snake_case_          =   'bird'
											snake_case_          =   jax.device_count()
											snake_case_          =   pipe.prepare_text_inputs([prompts] * num_samples	)

											snake_case_          =   load_image(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/bird_canny.png'	)
											snake_case_          =   pipe.prepare_image_inputs([canny_image] * num_samples	)

											snake_case_          =   jax.random.PRNGKey(0	)
											snake_case_          =   jax.random.split(snake_case				,       jax.device_count()	)

											snake_case_          =   replicate(snake_case	)
											snake_case_          =   shard(snake_case	)
											snake_case_          =   shard(snake_case	)

											snake_case_          =   pipe(
											    prompt_ids=snake_case				,       image=snake_case				,       params=snake_case				,       prng_seed=snake_case				,       num_inference_steps=50				,       jit=snake_case				,       ).images
											assert images.shape == (jax.device_count(), 1, 768, 512, 3)

											snake_case_          =   images.reshape((images.shape[0] * images.shape[1],) + images.shape[-3:]	)
											snake_case_          =   images[0, 253:256, 253:256, -1]

											snake_case_          =   jnp.asarray(jax.device_get(image_slice.flatten()	)	)
											snake_case_          =   jnp.array(
											    [0.16_79_69, 0.11_66_99, 0.08_15_43, 0.15_42_97, 0.13_28_12, 0.10_88_87, 0.16_99_22, 0.16_99_22, 0.20_50_78]	)
											print(F'''output_slice: {output_slice}'''	)
											assert jnp.abs(output_slice - expected_slice	).max() < 1e-2




				def    a							( self	):
											snake_case_    ,   snake_case_          =   FlaxControlNetModel.from_pretrained(
											    'lllyasviel/sd-controlnet-openpose'				,       from_pt=snake_case				,       dtype=jnp.bfloataa	)
											snake_case_    ,   snake_case_          =   FlaxStableDiffusionControlNetPipeline.from_pretrained(
											    'runwayml/stable-diffusion-v1-5'				,       controlnet=snake_case				,       from_pt=snake_case				,       dtype=jnp.bfloataa	)
											snake_case_          =   controlnet_params

											snake_case_          =   'Chef in the kitchen'
											snake_case_          =   jax.device_count()
											snake_case_          =   pipe.prepare_text_inputs([prompts] * num_samples	)

											snake_case_          =   load_image(
											    'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/sd_controlnet/pose.png'	)
											snake_case_          =   pipe.prepare_image_inputs([pose_image] * num_samples	)

											snake_case_          =   jax.random.PRNGKey(0	)
											snake_case_          =   jax.random.split(snake_case				,       jax.device_count()	)

											snake_case_          =   replicate(snake_case	)
											snake_case_          =   shard(snake_case	)
											snake_case_          =   shard(snake_case	)

											snake_case_          =   pipe(
											    prompt_ids=snake_case				,       image=snake_case				,       params=snake_case				,       prng_seed=snake_case				,       num_inference_steps=50				,       jit=snake_case				,       ).images
											assert images.shape == (jax.device_count(), 1, 768, 512, 3)

											snake_case_          =   images.reshape((images.shape[0] * images.shape[1],) + images.shape[-3:]	)
											snake_case_          =   images[0, 253:256, 253:256, -1]

											snake_case_          =   jnp.asarray(jax.device_get(image_slice.flatten()	)	)
											snake_case_          =   jnp.array(
											    [[0.27_14_84, 0.26_17_19, 0.27_53_91, 0.27_73_44, 0.27_92_97, 0.29_10_16, 0.29_49_22, 0.30_27_34, 0.30_27_34]]	)
											print(F'''output_slice: {output_slice}'''	)
											assert jnp.abs(output_slice - expected_slice	).max() < 1e-2




