

import collections.abc
from typing import Optional, Tuple, Union

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

from ...activations import ACTaFN
from ...modeling_outputs import BaseModelOutputWithNoAttention, ImageClassifierOutputWithNoAttention
from ...modeling_utils import PreTrainedModel
from ...utils import add_code_sample_docstrings, add_start_docstrings, add_start_docstrings_to_model_forward, logging
from .configuration_poolformer import PoolFormerConfig


A_	:			str  	=      logging.get_logger(__name__)

# General docstring
A_	:			Union[str, Any]  	=      'PoolFormerConfig'

# Base docstring
A_	:			Optional[Any]  	=      'sail/poolformer_s12'
A_	:			Any  	=      [1, 512, 7, 7]

# Image classification docstring
A_	:			List[str]  	=      'sail/poolformer_s12'
A_	:			Optional[Any]  	=      'tabby, tabby cat'

A_	:			List[str]  	=      [
    'sail/poolformer_s12',
    # See all PoolFormer models at https://huggingface.co/models?filter=poolformer
]

def 						UpperCamelCase (lowercase_:			Tuple   ,				lowercase_:			float = 0.0   ,				lowercase_:			bool = False )					->				List[str]:
	if drop_prob == 0.0 or not training:
		return input
	A__							:   List[Any]					       =		1 - drop_prob
	A__							:   List[str]					       =		(input.shape[0],) + (1,) * (input.ndim - 1)  # work with diff dim tensors, not just 2D ConvNets
	A__							:   str					       =		keep_prob + torch.rand(lowercase_   ,				dtype=input.dtype   ,				device=input.device )
	random_tensor.floor_()  # binarize
	A__							:   Optional[int]					       =		input.div(lowercase_ ) * random_tensor
	return output



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__ = None      ):
						super().__init__()
						A__							:   Dict					       =		drop_prob






					def   __A      (			self   ,	A__      ):
						return drop_path(A__   ,	self.drop_prob   ,	self.training      )






					def   __A      (			self      ):
						return "p={}".format(self.drop_prob      )



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__=None      ):
						super().__init__()
						A__							:   List[Any]					       =		patch_size if isinstance(A__   ,	collections.abc.Iterable      ) else (patch_size, patch_size)
						A__							:   Optional[Any]					       =		stride if isinstance(A__   ,	collections.abc.Iterable      ) else (stride, stride)
						A__							:   Optional[Any]					       =		padding if isinstance(A__   ,	collections.abc.Iterable      ) else (padding, padding)

						A__							:   Dict					       =		nn.Convad(A__   ,	A__   ,	kernel_size=A__   ,	stride=A__   ,	padding=A__      )
						A__							:   Optional[int]					       =		norm_layer(A__      ) if norm_layer else nn.Identity()






					def   __A      (			self   ,	A__      ):
						A__							:   Union[str, Any]					       =		self.projection(A__      )
						A__							:   Union[str, Any]					       =		self.norm(A__      )
						return embeddings



class    _a						(nn.GroupNorm    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	**A__      ):
						super().__init__(1   ,	A__   ,	**A__      )



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__      ):
						super().__init__()
						A__							:   List[str]					       =		nn.AvgPoolad(A__   ,	stride=1   ,	padding=pool_size // 2   ,	count_include_pad=A__      )






					def   __A      (			self   ,	A__      ):
						return self.pool(A__      ) - hidden_states



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__   ,	A__      ):
						super().__init__()
						A__							:   Tuple					       =		nn.Convad(A__   ,	A__   ,	1      )
						A__							:   Dict					       =		nn.Convad(A__   ,	A__   ,	1      )
						A__							:   int					       =		PoolFormerDropPath(A__      )
						if isinstance(config.hidden_act   ,	A__      ):
							A__							:   Dict					       =		ACTaFN[config.hidden_act]
						else:
							A__							:   str					       =		config.hidden_act






					def   __A      (			self   ,	A__      ):
						A__							:   int					       =		self.conva(A__      )
						A__							:   str					       =		self.act_fn(A__      )
						A__							:   Any					       =		self.drop(A__      )
						A__							:   str					       =		self.conva(A__      )
						A__							:   str					       =		self.drop(A__      )

						return hidden_states



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__   ,	A__      ):
						super().__init__()
						A__							:   Optional[Any]					       =		PoolFormerPooling(A__      )
						A__							:   List[Any]					       =		PoolFormerOutput(A__   ,	A__   ,	A__   ,	A__      )
						A__							:   int					       =		PoolFormerGroupNorm(A__      )
						A__							:   Optional[int]					       =		PoolFormerGroupNorm(A__      )

						# Useful for training neural nets
						A__							:   Any					       =		PoolFormerDropPath(A__      ) if drop_path > 0.0 else nn.Identity()
						A__							:   str					       =		config.use_layer_scale
						if config.use_layer_scale:
							A__							:   str					       =		nn.Parameter(
							    config.layer_scale_init_value * torch.ones((A__)      )   ,	requires_grad=A__      )
							A__							:   Any					       =		nn.Parameter(
							    config.layer_scale_init_value * torch.ones((A__)      )   ,	requires_grad=A__      )






					def   __A      (			self   ,	A__      ):
						if self.use_layer_scale:
							A__							:   str					       =		self.pooling(self.before_norm(A__      )      )
							A__							:   Any					       =		self.layer_scale_a.unsqueeze(-1      ).unsqueeze(-1      ) * pooling_output
							# First residual connection
							A__							:   Optional[int]					       =		hidden_states + self.drop_path(A__      )
							A__							:   Tuple					       =		()

							A__							:   Dict					       =		self.output(self.after_norm(A__      )      )
							A__							:   Union[str, Any]					       =		self.layer_scale_a.unsqueeze(-1      ).unsqueeze(-1      ) * layer_output
							# Second residual connection
							A__							:   Optional[int]					       =		hidden_states + self.drop_path(A__      )

							A__							:   int					       =		(output,) + outputs
							return outputs

						else:
							A__							:   List[str]					       =		self.drop_path(self.pooling(self.before_norm(A__      )      )      )
							# First residual connection
							A__							:   List[str]					       =		pooling_output + hidden_states
							A__							:   Optional[int]					       =		()

							# Second residual connection inside the PoolFormerOutput block
							A__							:   Tuple					       =		self.drop_path(self.output(self.after_norm(A__      )      )      )
							A__							:   List[str]					       =		hidden_states + layer_output

							A__							:   str					       =		(output,) + outputs
							return outputs



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__      ):
						super().__init__()
						A__							:   Any					       =		config
						# stochastic depth decay rule
						A__							:   List[Any]					       =		[x.item() for x in torch.linspace(0   ,	config.drop_path_rate   ,	sum(config.depths      )      )]

						# patch embeddings
						A__							:   List[Any]					       =		[]
						for i in range(config.num_encoder_blocks      ):
							embeddings.append(
							    PoolFormerEmbeddings(
							        patch_size=config.patch_sizes[i]   ,	stride=config.strides[i]   ,	padding=config.padding[i]   ,	num_channels=config.num_channels if i == 0 else config.hidden_sizes[i - 1]   ,	hidden_size=config.hidden_sizes[i]   ,	)      )
						A__							:   Any					       =		nn.ModuleList(A__      )

						# Transformer blocks
						A__							:   List[str]					       =		[]
						A__							:   Tuple					       =		0
						for i in range(config.num_encoder_blocks      ):
							# each block consists of layers
							A__							:   str					       =		[]
							if i != 0:
								cur += config.depths[i - 1]
							for j in range(config.depths[i]      ):
								layers.append(
								    PoolFormerLayer(
								        A__   ,	num_channels=config.hidden_sizes[i]   ,	pool_size=config.pool_size   ,	hidden_size=config.hidden_sizes[i]   ,	intermediate_size=int(config.hidden_sizes[i] * config.mlp_ratio      )   ,	drop_path=dpr[cur + j]   ,	)      )
							blocks.append(nn.ModuleList(A__      )      )

						A__							:   Dict					       =		nn.ModuleList(A__      )






					def   __A      (			self   ,	A__   ,	A__=False   ,	A__=True      ):
						A__							:   str					       =		() if output_hidden_states else None

						A__							:   List[str]					       =		pixel_values
						for idx, layers in enumerate(zip(self.patch_embeddings   ,	self.block      )      ):
							A__       ,       A__							:   Any					       =		layers
							# Get patch embeddings from hidden_states
							A__							:   Any					       =		embedding_layer(A__      )
							# Send the embeddings through the blocks
							for _, blk in enumerate(A__      ):
								A__							:   Optional[Any]					       =		blk(A__      )
								A__							:   int					       =		layer_outputs[0]

							if output_hidden_states:
								A__							:   Optional[Any]					       =		all_hidden_states + (hidden_states,)

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

						return BaseModelOutputWithNoAttention(last_hidden_state=A__   ,	hidden_states=A__      )



class    _a						(__magic_name__    ):


					'''simple docstring'''






					UpperCAmelCase__:       Union[str, Any]  							=	PoolFormerConfig
					UpperCAmelCase__:       Tuple  							=	'''poolformer'''
					UpperCAmelCase__:       Optional[int]  							=	'''pixel_values'''
					UpperCAmelCase__:       Optional[int]  							=	True






					def   __A      (			self   ,	A__      ):
						if isinstance(A__   ,	(nn.Linear, nn.Convad)      ):
							module.weight.data.normal_(mean=0.0   ,	std=self.config.initializer_range      )
							if module.bias is not None:
								module.bias.data.zero_()
						elif isinstance(A__   ,	nn.LayerNorm      ):
							module.bias.data.zero_()
							module.weight.data.fill_(1.0      )






					def   __A      (			self   ,	A__   ,	A__=False      ):
						if isinstance(A__   ,	A__      ):
							A__							:   List[Any]					       =		value



A_	:			Any  	=      r'\n    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) sub-class. Use\n    it as a regular PyTorch Module and refer to the PyTorch documentation for all matter related to general usage and\n    behavior.\n\n    Parameters:\n        config ([`PoolFormerConfig`]): 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 [`~PreTrainedModel.from_pretrained`] method to load the model weights.\n'

A_	:			Tuple  	=      r'\n    Args:\n        pixel_values (`torch.FloatTensor` of shape `(batch_size, num_channels, height, width)`):\n            Pixel values. Pixel values can be obtained using [`AutoImageProcessor`]. See\n            [`PoolFormerImageProcessor.__call__`] for details.\n'



@add_start_docstrings(
    '''The bare PoolFormer Model transformer outputting raw hidden-states without any specific head on top.'''		,       __magic_name__		,       )
class    _a						(__magic_name__    ):


					'''simple docstring'''






					def __init__(			self   ,	A__      ):
						super().__init__(A__      )
						A__							:   Dict					       =		config

						A__							:   Union[str, Any]					       =		PoolFormerEncoder(A__      )

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






					def   __A      (			self      ):
						return self.embeddings.patch_embeddings






					@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__ = None   ,	A__ = None   ,	A__ = None   ,	):
						A__							:   Tuple					       =		(
						    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
						)
						A__							:   Optional[Any]					       =		return_dict if return_dict is not None else self.config.use_return_dict

						if pixel_values is None:
							raise ValueError("""You have to specify pixel_values"""      )

						A__							:   str					       =		self.encoder(
						    A__   ,	output_hidden_states=A__   ,	return_dict=A__   ,	)
						A__							:   Tuple					       =		encoder_outputs[0]

						if not return_dict:
							return (sequence_output, None) + encoder_outputs[1:]

						return BaseModelOutputWithNoAttention(
						    last_hidden_state=A__   ,	hidden_states=encoder_outputs.hidden_states   ,	)



class    _a						(nn.Module    ):


					'''simple docstring'''






					def __init__(			self   ,	A__      ):
						super().__init__()
						A__							:   Tuple					       =		nn.Linear(config.hidden_size   ,	config.hidden_size      )






					def   __A      (			self   ,	A__      ):
						A__							:   Optional[int]					       =		self.dense(A__      )
						return output



@add_start_docstrings(
    '''
    PoolFormer Model transformer with an image classification head on top
    '''		,       __magic_name__		,       )
class    _a						(__magic_name__    ):


					'''simple docstring'''






					def __init__(			self   ,	A__      ):
						super().__init__(A__      )
						A__							:   Optional[int]					       =		config.num_labels
						A__							:   Union[str, Any]					       =		PoolFormerModel(A__      )

						# Final norm
						A__							:   str					       =		PoolFormerGroupNorm(config.hidden_sizes[-1]      )
						# Classifier head
						A__							:   List[str]					       =		(
						    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__							:   Optional[int]					       =		return_dict if return_dict is not None else self.config.use_return_dict

						A__							:   str					       =		self.poolformer(
						    A__   ,	output_hidden_states=A__   ,	return_dict=A__   ,	)

						A__							:   str					       =		outputs[0]

						A__							:   Tuple					       =		self.classifier(self.norm(A__      ).mean([-2, -1]      )      )

						A__							:   Optional[Any]					       =		None
						if labels is not None:
							if self.config.problem_type is None:
								if self.num_labels == 1:
									A__							:   Dict					       =		"""regression"""
								elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int):
									A__							:   Union[str, Any]					       =		"""single_label_classification"""
								else:
									A__							:   int					       =		"""multi_label_classification"""

							if self.config.problem_type == "regression":
								A__							:   Union[str, Any]					       =		MSELoss()
								if self.num_labels == 1:
									A__							:   int					       =		loss_fct(logits.squeeze()   ,	labels.squeeze()      )
								else:
									A__							:   str					       =		loss_fct(A__   ,	A__      )
							elif self.config.problem_type == "single_label_classification":
								A__							:   List[Any]					       =		CrossEntropyLoss()
								A__							:   List[str]					       =		loss_fct(logits.view(-1   ,	self.num_labels      )   ,	labels.view(-1      )      )
							elif self.config.problem_type == "multi_label_classification":
								A__							:   Optional[int]					       =		BCEWithLogitsLoss()
								A__							:   str					       =		loss_fct(A__   ,	A__      )

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

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






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

import numpy as np
from huggingface_hub import hf_hub_download

from ..utils import ONNX_EXTERNAL_WEIGHTS_NAME, ONNX_WEIGHTS_NAME, is_onnx_available, logging


if is_onnx_available():
	import onnxruntime as ort


A_	:			Union[str, Any]  	=      logging.get_logger(__name__)

A_	:			str  	=      {
    'tensor(bool)': np.bool_,
    'tensor(int8)': np.inta,
    'tensor(uint8)': np.uinta,
    'tensor(int16)': np.intaa,
    'tensor(uint16)': np.uintaa,
    'tensor(int32)': np.intaa,
    'tensor(uint32)': np.uintaa,
    'tensor(int64)': np.intaa,
    'tensor(uint64)': np.uintaa,
    'tensor(float16)': np.floataa,
    'tensor(float)': np.floataa,
    'tensor(double)': np.floataa,
}



class    _a						:


					'''simple docstring'''






					def __init__(			self   ,	A__=None   ,	**A__      ):
						logger.info("""`diffusers.OnnxRuntimeModel` is experimental and might change in the future."""      )
						A__							:   Dict					       =		model
						A__							:   Any					       =		kwargs.get("""model_save_dir"""   ,	A__      )
						A__							:   Optional[int]					       =		kwargs.get("""latest_model_name"""   ,	A__      )






					def __call__(			self   ,	**A__      ):
						A__							:   int					       =		{k: np.array(A__      ) for k, v in kwargs.items()}
						return self.model.run(A__   ,	A__      )






					@staticmethod
					def   __A      (			A__   ,	A__=None   ,	A__=None      ):
						if provider is None:
							logger.info("""No onnxruntime provider specified, using CPUExecutionProvider"""      )
							A__							:   List[Any]					       =		"""CPUExecutionProvider"""

						return ort.InferenceSession(A__   ,	providers=[provider]   ,	sess_options=A__      )






					def   __A      (			self   ,	A__   ,	A__ = None   ,	**A__      ):
						A__							:   List[Any]					       =		file_name if file_name is not None else ONNX_WEIGHTS_NAME

						A__							:   List[Any]					       =		self.model_save_dir.joinpath(self.latest_model_name      )
						A__							:   Optional[int]					       =		Path(A__      ).joinpath(A__      )
						try:
							shutil.copyfile(A__   ,	A__      )
						except shutil.SameFileError:
							pass

						# copy external weights (for models >2GB)
						A__							:   str					       =		self.model_save_dir.joinpath(A__      )
						if src_path.exists():
							A__							:   List[str]					       =		Path(A__      ).joinpath(A__      )
							try:
								shutil.copyfile(A__   ,	A__      )
							except shutil.SameFileError:
								pass






					def   __A      (			self   ,	A__   ,	**A__   ,	):
						if os.path.isfile(A__      ):
							logger.error(F"""Provided path ({save_directory}) should be a directory, not a file"""      )
							return

						os.makedirs(A__   ,	exist_ok=A__      )

						# saving model weights/files
						self._save_pretrained(A__   ,	**A__      )






					@classmethod
					def   __A      (			cls   ,	A__   ,	A__ = None   ,	A__ = None   ,	A__ = False   ,	A__ = None   ,	A__ = None   ,	A__ = None   ,	A__ = None   ,	**A__   ,	):
						A__							:   str					       =		file_name if file_name is not None else ONNX_WEIGHTS_NAME
						# load model from local directory
						if os.path.isdir(A__      ):
							A__							:   Dict					       =		OnnxRuntimeModel.load_model(
							    os.path.join(A__   ,	A__      )   ,	provider=A__   ,	sess_options=A__      )
							A__							:   Optional[Any]					       =		Path(A__      )
						# load model from hub
						else:
							# download model
							A__							:   Union[str, Any]					       =		hf_hub_download(
							    repo_id=A__   ,	filename=A__   ,	use_auth_token=A__   ,	revision=A__   ,	cache_dir=A__   ,	force_download=A__   ,	)
							A__							:   List[str]					       =		Path(A__      ).parent
							A__							:   str					       =		Path(A__      ).name
							A__							:   Optional[int]					       =		OnnxRuntimeModel.load_model(A__   ,	provider=A__   ,	sess_options=A__      )
						return cls(model=A__   ,	**A__      )






					@classmethod
					def   __A      (			cls   ,	A__   ,	A__ = True   ,	A__ = None   ,	A__ = None   ,	**A__   ,	):
						A__							:   Optional[Any]					       =		None
						if len(str(A__      ).split("""@"""      )      ) == 2:
							A__       ,       A__							:   Union[str, Any]					       =		model_id.split("""@"""      )

						return cls._from_pretrained(
						    model_id=A__   ,	revision=A__   ,	cache_dir=A__   ,	force_download=A__   ,	use_auth_token=A__   ,	**A__   ,	)



