



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							 =					logging.get_logger(__name__)

# General docstring
UpperCAmelCase							 =					"""ResNetConfig"""

# Base docstring
UpperCAmelCase							 =					"""microsoft/resnet-50"""
UpperCAmelCase							 =					[1, 2_048, 7, 7]

# Image classification docstring
UpperCAmelCase							 =					"""microsoft/resnet-50"""
UpperCAmelCase							 =					"""tiger cat"""

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

class  __snake_case      (   nn.Module):







							'''simple docstring'''






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







							def _a      (							self  ,  a_		):
											a__				      =		self.convolution(a_		)
											a__				      =		self.normalization(a_		)
											a__				      =		self.activation(a_		)
											return hidden_state




class  __snake_case      (   nn.Module):







							'''simple docstring'''






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







							def _a      (							self  ,  a_		):
											a__				      =		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."""		)
											a__				      =		self.embedder(a_		)
											a__				      =		self.pooler(a_		)
											return embedding




class  __snake_case      (   nn.Module):







							'''simple docstring'''






							def __init__(							self  ,  a_  ,  a_  ,  a_ = 2		):
											super().__init__()
											a__				      =		nn.Convad(a_  ,  a_  ,  kernel_size=1  ,  stride=a_  ,  bias=a_		)
											a__				      =		nn.BatchNormad(a_		)







							def _a      (							self  ,  a_		):
											a__				      =		self.convolution(a_		)
											a__				      =		self.normalization(a_		)
											return hidden_state




class  __snake_case      (   nn.Module):







							'''simple docstring'''






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







							def _a      (							self  ,  a_		):
											a__				      =		hidden_state
											a__				      =		self.layer(a_		)
											a__				      =		self.shortcut(a_		)
											hidden_state += residual
											a__				      =		self.activation(a_		)
											return hidden_state




class  __snake_case      (   nn.Module):







							'''simple docstring'''






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







							def _a      (							self  ,  a_		):
											a__				      =		hidden_state
											a__				      =		self.layer(a_		)
											a__				      =		self.shortcut(a_		)
											hidden_state += residual
											a__				      =		self.activation(a_		)
											return hidden_state




class  __snake_case      (   nn.Module):







							'''simple docstring'''






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

											a__				      =		ResNetBottleNeckLayer if config.layer_type == """bottleneck""" else ResNetBasicLayer

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







							def _a      (							self  ,  a_		):
											a__				      =		input
											for layer in self.layers:
															a__				      =		layer(a_		)
											return hidden_state




class  __snake_case      (   nn.Module):







							'''simple docstring'''






							def __init__(							self  ,  a_		):
											super().__init__()
											a__				      =		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(
											        a_  ,  config.embedding_size  ,  config.hidden_sizes[0]  ,  stride=2 if config.downsample_in_first_stage else 1  ,  depth=config.depths[0]  ,  )		)
											a__				      =		zip(config.hidden_sizes  ,  config.hidden_sizes[1:]		)
											for (in_channels, out_channels), depth in zip(a_  ,  config.depths[1:]		):
															self.stages.append(ResNetStage(a_  ,  a_  ,  a_  ,  depth=a_		)		)







							def _a      (							self  ,  a_  ,  a_ = False  ,  a_ = True		):
											a__				      =		() if output_hidden_states else None

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

															a__				      =		stage_module(a_		)

											if output_hidden_states:
															a__				      =		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=a_  ,  hidden_states=a_  ,  )




class  __snake_case      (   SCREAMING_SNAKE_CASE):







							'''simple docstring'''






							UpperCamelCase__				:      Union[str, Any]         =					ResNetConfig
							UpperCamelCase__				:      Union[str, Any]         =					"""resnet"""
							UpperCamelCase__				:      Optional[int]         =					"""pixel_values"""
							UpperCamelCase__				:      Tuple         =					True


							def _a      (							self  ,  a_		):
											if isinstance(a_  ,  nn.Convad		):
															nn.init.kaiming_normal_(module.weight  ,  mode="""fan_out"""  ,  nonlinearity="""relu"""		)
											elif isinstance(a_  ,  (nn.BatchNormad, nn.GroupNorm)		):
															nn.init.constant_(module.weight  ,  1		)
															nn.init.constant_(module.bias  ,  0		)







							def _a      (							self  ,  a_  ,  a_=False		):
											if isinstance(a_  ,  a_		):
															a__				      =		value




UpperCAmelCase							 =					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							 =					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."""			,SCREAMING_SNAKE_CASE			,)
class  __snake_case      (   SCREAMING_SNAKE_CASE):







							'''simple docstring'''






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







							@add_start_docstrings_to_model_forward(a_		)
							@add_code_sample_docstrings(
							    checkpoint=_CHECKPOINT_FOR_DOC  ,  output_type=a_  ,  config_class=_CONFIG_FOR_DOC  ,  modality="""vision"""  ,  expected_output=_EXPECTED_OUTPUT_SHAPE  ,  )
							def _a      (							self  ,  a_  ,  a_ = None  ,  a_ = None		):
											a__				      =		(
											    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
											)
											a__				      =		return_dict if return_dict is not None else self.config.use_return_dict

											a__				      =		self.embedder(a_		)

											a__				      =		self.encoder(
											    a_  ,  output_hidden_states=a_  ,  return_dict=a_		)

											a__				      =		encoder_outputs[0]

											a__				      =		self.pooler(a_		)

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

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




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







							'''simple docstring'''






							def __init__(							self  ,  a_		):
											super().__init__(a_		)
											a__				      =		config.num_labels
											a__				      =		ResNetModel(a_		)
											# classification head
											a__				      =		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(a_		)
							@add_code_sample_docstrings(
							    checkpoint=_IMAGE_CLASS_CHECKPOINT  ,  output_type=a_  ,  config_class=_CONFIG_FOR_DOC  ,  expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT  ,  )
							def _a      (							self  ,  a_ = None  ,  a_ = None  ,  a_ = None  ,  a_ = None  ,  ):
											a__				      =		return_dict if return_dict is not None else self.config.use_return_dict

											a__				      =		self.resnet(a_  ,  output_hidden_states=a_  ,  return_dict=a_		)

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

											a__				      =		self.classifier(a_		)

											a__				      =		None

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

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

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




@add_start_docstrings(
    """
    ResNet backbone, to be used with frameworks like DETR and MaskFormer.
    """			,SCREAMING_SNAKE_CASE			,)
class  __snake_case      (   SCREAMING_SNAKE_CASE			,SCREAMING_SNAKE_CASE):







							'''simple docstring'''






							def __init__(							self  ,  a_		):
											super().__init__(a_		)
											super()._init_backbone(a_		)

											a__				      =		[config.embedding_size] + config.hidden_sizes
											a__				      =		ResNetEmbeddings(a_		)
											a__				      =		ResNetEncoder(a_		)

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







							@add_start_docstrings_to_model_forward(a_		)
							@replace_return_docstrings(output_type=a_  ,  config_class=_CONFIG_FOR_DOC		)
							def _a      (							self  ,  a_  ,  a_ = None  ,  a_ = None		):
											a__				      =		return_dict if return_dict is not None else self.config.use_return_dict
											a__				      =		(
											    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
											)

											a__				      =		self.embedder(a_		)

											a__				      =		self.encoder(a_  ,  output_hidden_states=a_  ,  return_dict=a_		)

											a__				      =		outputs.hidden_states

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

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

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






import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


UpperCAmelCase							 =					logging.get_logger(__name__)

UpperCAmelCase							 =					{"""vocab_file""": """spiece.model"""}

UpperCAmelCase							 =					{
    """vocab_file""": {
        """bert_for_seq_generation""": (
            """https://huggingface.co/google/bert_for_seq_generation_L-24_bbc_encoder/resolve/main/spiece.model"""
        ),
    }
}

UpperCAmelCase							 =					{"""bert_for_seq_generation""": 512}

class  __snake_case      (   SCREAMING_SNAKE_CASE):







							'''simple docstring'''






							UpperCamelCase__				:      Optional[Any]         =					VOCAB_FILES_NAMES
							UpperCamelCase__				:      Any         =					PRETRAINED_VOCAB_FILES_MAP
							UpperCamelCase__				:      List[str]         =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							UpperCamelCase__				:      List[int]         =					[]
							UpperCamelCase__				:      Optional[int]         =					["""input_ids""", """attention_mask"""]


							def __init__(							self  ,  a_  ,  a_="<s>"  ,  a_="</s>"  ,  a_="<unk>"  ,  a_="<pad>"  ,  a_="<::::>"  ,  a_ = None  ,  **a_  ,  ):
											a__				      =		{} if sp_model_kwargs is None else sp_model_kwargs

											# Add extra_ids to the special token list
											super().__init__(
											    bos_token=a_  ,  eos_token=a_  ,  unk_token=a_  ,  pad_token=a_  ,  sep_token=a_  ,  sp_model_kwargs=self.sp_model_kwargs  ,  **a_  ,  )

											a__				      =		vocab_file

											a__				      =		spm.SentencePieceProcessor(**self.sp_model_kwargs		)
											self.sp_model.Load(a_		)


							@property
							def _a      (							self		):
											return self.sp_model.get_piece_size()


							def _a      (							self		):
											a__				      =		{self.convert_ids_to_tokens(a_		): i for i in range(self.vocab_size		)}
											vocab.update(self.added_tokens_encoder		)
											return vocab


							def __getstate__(							self		):
											a__				      =		self.__dict__.copy()
											a__				      =		None
											return state


							def __setstate__(							self  ,  a_		):
											a__				      =		d

											# for backward compatibility
											if not hasattr(self  ,  """sp_model_kwargs"""		):
															a__				      =		{}

											a__				      =		spm.SentencePieceProcessor(**self.sp_model_kwargs		)
											self.sp_model.Load(self.vocab_file		)


							def _a      (							self  ,  a_		):
											return self.sp_model.encode(a_  ,  out_type=a_		)


							def _a      (							self  ,  a_		):
											return self.sp_model.piece_to_id(a_		)


							def _a      (							self  ,  a_		):
											a__				      =		self.sp_model.IdToPiece(a_		)
											return token


							def _a      (							self  ,  a_		):
											a__				      =		[]
											a__				      =		""""""
											for token in tokens:
															# make sure that special tokens are not decoded using sentencepiece model
															if token in self.all_special_tokens:
																			out_string += self.sp_model.decode(a_		) + token
																			a__				      =		[]
															else:
																			current_sub_tokens.append(a_		)
											out_string += self.sp_model.decode(a_		)
											return out_string.strip()







							def _a      (							self  ,  a_  ,  a_ = None		):
											if not os.path.isdir(a_		):
															logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''		)
															return
											a__				      =		os.path.join(
											    a_  ,  (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)

											if os.path.abspath(self.vocab_file		) != os.path.abspath(a_		) and os.path.isfile(self.vocab_file		):
															copyfile(self.vocab_file  ,  a_		)
											elif not os.path.isfile(self.vocab_file		):
															with open(a_  ,  """wb"""		) as fi:
																			a__				      =		self.sp_model.serialized_model_proto()
																			fi.write(a_		)

											return (out_vocab_file,)

