


"""simple docstring"""






from typing import Optional, Union

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

from ...activations import ACTaFN
from ...modeling_outputs import BaseModelOutputWithPoolingAndNoAttention, 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_mobilenet_va import MobileNetVaConfig


_lowercase			:			Any										=						logging.get_logger(__name__)


# General docstring
_lowercase			:			Any										=						"MobileNetV1Config"

# Base docstring
_lowercase			:			int										=						"google/mobilenet_v1_1.0_224"
_lowercase			:			Tuple										=						[1, 1_0_2_4, 7, 7]

# Image classification docstring
_lowercase			:			Tuple										=						"google/mobilenet_v1_1.0_224"
_lowercase			:			str										=						"tabby, tabby cat"


_lowercase			:			Optional[Any]										=						[
    "google/mobilenet_v1_1.0_224",
    "google/mobilenet_v1_0.75_192",
    # See all MobileNetV1 models at https://huggingface.co/models?filter=mobilenet_v1
]






def 					snake_case__       (		__lowerCamelCase    :   List[Any]							,		__lowerCamelCase    :   int							,		__lowerCamelCase    :   str=None     ):
			"""simple docstring"""
			lowerCamelCase__		:		int		     ={}

			if isinstance(__lowerCamelCase							,		__lowerCamelCase     ):
						lowerCamelCase__		:		int		     =model.mobilenet_va
			else:
						lowerCamelCase__		:		Optional[int]		     =model

			lowerCamelCase__		:		Dict		     ='''MobilenetV1/Conv2d_0/'''
			lowerCamelCase__		:		List[str]		     =backbone.conv_stem.convolution.weight
			lowerCamelCase__		:		List[str]		     =backbone.conv_stem.normalization.bias
			lowerCamelCase__		:		Dict		     =backbone.conv_stem.normalization.weight
			lowerCamelCase__		:		Optional[Any]		     =backbone.conv_stem.normalization.running_mean
			lowerCamelCase__		:		List[str]		     =backbone.conv_stem.normalization.running_var

			for i in range(13     ):
						lowerCamelCase__		:		int		     =i + 1
						lowerCamelCase__		:		Optional[int]		     =i * 2

						lowerCamelCase__		:		str		     =backbone.layer[pt_index]
						lowerCamelCase__		:		List[Any]		     =f'''MobilenetV1/Conv2d_{tf_index}_depthwise/'''
						lowerCamelCase__		:		Dict		     =pointer.convolution.weight
						lowerCamelCase__		:		Optional[int]		     =pointer.normalization.bias
						lowerCamelCase__		:		List[Any]		     =pointer.normalization.weight
						lowerCamelCase__		:		List[str]		     =pointer.normalization.running_mean
						lowerCamelCase__		:		Dict		     =pointer.normalization.running_var

						lowerCamelCase__		:		Optional[Any]		     =backbone.layer[pt_index + 1]
						lowerCamelCase__		:		int		     =f'''MobilenetV1/Conv2d_{tf_index}_pointwise/'''
						lowerCamelCase__		:		List[Any]		     =pointer.convolution.weight
						lowerCamelCase__		:		Union[str, Any]		     =pointer.normalization.bias
						lowerCamelCase__		:		Tuple		     =pointer.normalization.weight
						lowerCamelCase__		:		Union[str, Any]		     =pointer.normalization.running_mean
						lowerCamelCase__		:		int		     =pointer.normalization.running_var

			if isinstance(__lowerCamelCase							,		__lowerCamelCase     ):
						lowerCamelCase__		:		Dict		     ='''MobilenetV1/Logits/Conv2d_1c_1x1/'''
						lowerCamelCase__		:		str		     =model.classifier.weight
						lowerCamelCase__		:		Any		     =model.classifier.bias

			return tf_to_pt_map






def 					snake_case__       (		__lowerCamelCase    :   Optional[Any]							,		__lowerCamelCase    :   str							,		__lowerCamelCase    :   Optional[int]     ):
			"""simple docstring"""
			try:
						import numpy as np
						import tensorflow as tf
			except ImportError:
						logger.error(
						    '''Loading a TensorFlow models in PyTorch, requires TensorFlow to be installed. Please see '''
						    '''https://www.tensorflow.org/install/ for installation instructions.'''     )
						raise

			# Load weights from TF model
			lowerCamelCase__		:		int		     =tf.train.list_variables(__lowerCamelCase     )
			lowerCamelCase__		:		Tuple		     ={}
			for name, shape in init_vars:
						logger.info(f'''Loading TF weight {name} with shape {shape}'''     )
						lowerCamelCase__		:		Any		     =tf.train.load_variable(__lowerCamelCase							,		__lowerCamelCase     )
						lowerCamelCase__		:		Optional[Any]		     =array

			# Build TF to PyTorch weights loading map
			lowerCamelCase__		:		Optional[int]		     =_build_tf_to_pytorch_map(__lowerCamelCase							,		__lowerCamelCase							,		__lowerCamelCase     )

			for name, pointer in tf_to_pt_map.items():
						logger.info(f'''Importing {name}'''     )
						if name not in tf_weights:
									logger.info(f'''{name} not in tf pre-trained weights, skipping'''     )
									continue

						lowerCamelCase__		:		int		     =tf_weights[name]

						if "depthwise_weights" in name:
									logger.info('''Transposing depthwise'''     )
									lowerCamelCase__		:		Any		     =np.transpose(__lowerCamelCase							,		(2, 3, 0, 1)     )
						elif "weights" in name:
									logger.info('''Transposing'''     )
									if len(pointer.shape     ) == 2:  # copying into linear layer
												lowerCamelCase__		:		Any		     =array.squeeze().transpose()
									else:
												lowerCamelCase__		:		Any		     =np.transpose(__lowerCamelCase							,		(3, 2, 0, 1)     )

						if pointer.shape != array.shape:
									raise ValueError(f'''Pointer shape {pointer.shape} and array shape {array.shape} mismatched'''     )

						logger.info(f'''Initialize PyTorch weight {name} {array.shape}'''     )
						lowerCamelCase__		:		Optional[int]		     =torch.from_numpy(__lowerCamelCase     )

						tf_weights.pop(__lowerCamelCase							,		__lowerCamelCase     )
						tf_weights.pop(name + '''/RMSProp'''							,		__lowerCamelCase     )
						tf_weights.pop(name + '''/RMSProp_1'''							,		__lowerCamelCase     )
						tf_weights.pop(name + '''/ExponentialMovingAverage'''							,		__lowerCamelCase     )

			logger.info(f'''Weights not copied to PyTorch model: {", ".join(tf_weights.keys()     )}'''     )
			return model






def 					snake_case__       (		__lowerCamelCase    :   Optional[Any]							,		__lowerCamelCase    :   Optional[Any]     ):
			"""simple docstring"""
			lowerCamelCase__     ,       lowerCamelCase__		:		Union[str, Any]		     =features.shape[-2:]
			lowerCamelCase__     ,       lowerCamelCase__		:		Tuple		     =conv_layer.stride
			lowerCamelCase__     ,       lowerCamelCase__		:		Optional[int]		     =conv_layer.kernel_size

			if in_height % stride_height == 0:
						lowerCamelCase__		:		List[Any]		     =max(kernel_height - stride_height							,		0     )
			else:
						lowerCamelCase__		:		str		     =max(kernel_height - (in_height % stride_height)							,		0     )

			if in_width % stride_width == 0:
						lowerCamelCase__		:		List[Any]		     =max(kernel_width - stride_width							,		0     )
			else:
						lowerCamelCase__		:		Union[str, Any]		     =max(kernel_width - (in_width % stride_width)							,		0     )

			lowerCamelCase__		:		Dict		     =pad_along_width // 2
			lowerCamelCase__		:		int		     =pad_along_width - pad_left
			lowerCamelCase__		:		Tuple		     =pad_along_height // 2
			lowerCamelCase__		:		Optional[Any]		     =pad_along_height - pad_top

			lowerCamelCase__		:		List[str]		     =(pad_left, pad_right, pad_top, pad_bottom)
			return nn.functional.pad(__lowerCamelCase							,		__lowerCamelCase							,		'''constant'''							,		0.0     )

class 	__SCREAMING_SNAKE_CASE     (				nn.Module       ):


					'''simple docstring'''





					def __init__(	self							:  Union[str, Any],  lowerCamelCase							:  MobileNetVaConfig,  lowerCamelCase							:  int,  lowerCamelCase							:  int,  lowerCamelCase							:  int,  lowerCamelCase							:  Optional[int] = 1,  lowerCamelCase							:  Optional[int] = 1,  lowerCamelCase							:  bool = False,  lowerCamelCase							:  Optional[bool] = True,  lowerCamelCase							:  Optional[bool or str] = True,  )-> Any:
								super().__init__()
								lowerCamelCase__		:		Optional[Any]		     =config

								if in_channels % groups != 0:
											raise ValueError(F'''Input channels ({in_channels}) are not divisible by {groups} groups.'''					)
								if out_channels % groups != 0:
											raise ValueError(F'''Output channels ({out_channels}) are not divisible by {groups} groups.'''					)

								lowerCamelCase__		:		Optional[int]		     =0 if config.tf_padding else int((kernel_size - 1) / 2					)

								lowerCamelCase__		:		List[str]		     =nn.Convad(
								    in_channels=__lowerCAmelCase,  out_channels=__lowerCAmelCase,  kernel_size=__lowerCAmelCase,  stride=__lowerCAmelCase,  padding=__lowerCAmelCase,  groups=__lowerCAmelCase,  bias=__lowerCAmelCase,  padding_mode='''zeros''',  )

								if use_normalization:
											lowerCamelCase__		:		Optional[int]		     =nn.BatchNormad(
											    num_features=__lowerCAmelCase,  eps=config.layer_norm_eps,  momentum=0.9_997,  affine=__lowerCAmelCase,  track_running_stats=__lowerCAmelCase,  )
								else:
											lowerCamelCase__		:		List[Any]		     =None

								if use_activation:
											if isinstance(__lowerCAmelCase,  __lowerCAmelCase					):
														lowerCamelCase__		:		List[Any]		     =ACTaFN[use_activation]
											elif isinstance(config.hidden_act,  __lowerCAmelCase					):
														lowerCamelCase__		:		Any		     =ACTaFN[config.hidden_act]
											else:
														lowerCamelCase__		:		Union[str, Any]		     =config.hidden_act
								else:
											lowerCamelCase__		:		Optional[int]		     =None




					def 		snake_case	(	self							:  Optional[Any],  lowerCamelCase							:  torch.Tensor					)-> Optional[int]:
								if self.config.tf_padding:
											lowerCamelCase__		:		int		     =apply_tf_padding(__lowerCAmelCase,  self.convolution					)
								lowerCamelCase__		:		List[Any]		     =self.convolution(__lowerCAmelCase					)
								if self.normalization is not None:
											lowerCamelCase__		:		List[Any]		     =self.normalization(__lowerCAmelCase					)
								if self.activation is not None:
											lowerCamelCase__		:		Tuple		     =self.activation(__lowerCAmelCase					)
								return features





class 	__SCREAMING_SNAKE_CASE     (				lowerCAmelCase_       ):


					'''simple docstring'''





					_a             =							MobileNetVaConfig
					_a             =							load_tf_weights_in_mobilenet_va
					_a             =							'mobilenet_v1'
					_a             =							'pixel_values'
					_a             =							False



					def 		snake_case	(	self							:  int,  lowerCamelCase							:  Union[nn.Linear, nn.Convad]					)-> List[Any]:
								if isinstance(__lowerCAmelCase,  (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(__lowerCAmelCase,  nn.BatchNormad					):
											module.bias.data.zero_()
											module.weight.data.fill_(1.0					)





_lowercase			:			Tuple										=						r"\n    This model is a PyTorch [torch.nn.Module](https://pytorch.org/docs/stable/nn.html#torch.nn.Module) subclass. Use it\n    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 ([`MobileNetV1Config`]): 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"

_lowercase			:			Optional[int]										=						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            [`MobileNetV1ImageProcessor.__call__`] for details.\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"

@add_start_docstrings(
    'The bare MobileNetV1 model outputting raw hidden-states without any specific head on top.'		,       lowerCAmelCase_		,       )
class 	__SCREAMING_SNAKE_CASE     (				lowerCAmelCase_       ):


					'''simple docstring'''





					def __init__(	self							:  str,  lowerCamelCase							:  MobileNetVaConfig,  lowerCamelCase							:  bool = True					)-> Any:
								super().__init__(__lowerCAmelCase					)
								lowerCamelCase__		:		int		     =config

								lowerCamelCase__		:		int		     =32
								lowerCamelCase__		:		Union[str, Any]		     =max(int(depth * config.depth_multiplier					),  config.min_depth					)

								lowerCamelCase__		:		List[str]		     =MobileNetVaConvLayer(
								    __lowerCAmelCase,  in_channels=config.num_channels,  out_channels=__lowerCAmelCase,  kernel_size=3,  stride=2,  )

								lowerCamelCase__		:		Optional[int]		     =[1, 2, 1, 2, 1, 2, 1, 1, 1, 1, 1, 2, 1]

								lowerCamelCase__		:		Tuple		     =nn.ModuleList()
								for i in range(13					):
											lowerCamelCase__		:		Union[str, Any]		     =out_channels

											if strides[i] == 2 or i == 0:
														depth *= 2
														lowerCamelCase__		:		Any		     =max(int(depth * config.depth_multiplier					),  config.min_depth					)

											self.layer.append(
											    MobileNetVaConvLayer(
											        __lowerCAmelCase,  in_channels=__lowerCAmelCase,  out_channels=__lowerCAmelCase,  kernel_size=3,  stride=strides[i],  groups=__lowerCAmelCase,  )					)

											self.layer.append(
											    MobileNetVaConvLayer(
											        __lowerCAmelCase,  in_channels=__lowerCAmelCase,  out_channels=__lowerCAmelCase,  kernel_size=1,  )					)

								lowerCamelCase__		:		str		     =nn.AdaptiveAvgPoolad((1, 1)					) if add_pooling_layer else None

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



					def 		snake_case	(	self							:  Any,  lowerCamelCase							:  Union[str, Any]					)-> Any:
								raise NotImplementedError




					@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 		snake_case	(	self							:  Dict,  lowerCamelCase							:  Optional[torch.Tensor] = None,  lowerCamelCase							:  Optional[bool] = None,  lowerCamelCase							:  Optional[bool] = None,  )-> Union[str, Any]:
								lowerCamelCase__		:		Tuple		     =(
								    output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
								)
								lowerCamelCase__		:		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'''					)

								lowerCamelCase__		:		Union[str, Any]		     =self.conv_stem(__lowerCAmelCase					)

								lowerCamelCase__		:		List[str]		     =() if output_hidden_states else None

								for i, layer_module in enumerate(self.layer					):
											lowerCamelCase__		:		List[str]		     =layer_module(__lowerCAmelCase					)

											if output_hidden_states:
														lowerCamelCase__		:		Dict		     =all_hidden_states + (hidden_states,)

								lowerCamelCase__		:		Optional[int]		     =hidden_states

								if self.pooler is not None:
											lowerCamelCase__		:		Dict		     =torch.flatten(self.pooler(__lowerCAmelCase					),  start_dim=1					)
								else:
											lowerCamelCase__		:		Dict		     =None

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

								return BaseModelOutputWithPoolingAndNoAttention(
								    last_hidden_state=__lowerCAmelCase,  pooler_output=__lowerCAmelCase,  hidden_states=__lowerCAmelCase,  )





@add_start_docstrings(
    '\n    MobileNetV1 model with an image classification head on top (a linear layer on top of the pooled features), e.g. for\n    ImageNet.\n    '		,       lowerCAmelCase_		,       )
class 	__SCREAMING_SNAKE_CASE     (				lowerCAmelCase_       ):


					'''simple docstring'''





					def __init__(	self							:  Optional[int],  lowerCamelCase							:  MobileNetVaConfig					)-> Optional[int]:
								super().__init__(__lowerCAmelCase					)

								lowerCamelCase__		:		Tuple		     =config.num_labels
								lowerCamelCase__		:		int		     =MobileNetVaModel(__lowerCAmelCase					)

								lowerCamelCase__		:		Optional[Any]		     =self.mobilenet_va.layer[-1].convolution.out_channels

								# Classifier head
								lowerCamelCase__		:		List[str]		     =nn.Dropout(config.classifier_dropout_prob,  inplace=__lowerCAmelCase					)
								lowerCamelCase__		:		List[str]		     =nn.Linear(__lowerCAmelCase,  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 		snake_case	(	self							:  Optional[Any],  lowerCamelCase							:  Optional[torch.Tensor] = None,  lowerCamelCase							:  Optional[bool] = None,  lowerCamelCase							:  Optional[torch.Tensor] = None,  lowerCamelCase							:  Optional[bool] = None,  )-> int:
								lowerCamelCase__		:		int		     =return_dict if return_dict is not None else self.config.use_return_dict

								lowerCamelCase__		:		Any		     =self.mobilenet_va(__lowerCAmelCase,  output_hidden_states=__lowerCAmelCase,  return_dict=__lowerCAmelCase					)

								lowerCamelCase__		:		Tuple		     =outputs.pooler_output if return_dict else outputs[1]

								lowerCamelCase__		:		int		     =self.classifier(self.dropout(__lowerCAmelCase					)					)

								lowerCamelCase__		:		Any		     =None
								if labels is not None:
											if self.config.problem_type is None:
														if self.num_labels == 1:
																	lowerCamelCase__		:		int		     ='''regression'''
														elif self.num_labels > 1 and (labels.dtype == torch.long or labels.dtype == torch.int):
																	lowerCamelCase__		:		int		     ='''single_label_classification'''
														else:
																	lowerCamelCase__		:		Dict		     ='''multi_label_classification'''

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

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

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









"""simple docstring"""






import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModelWithProjection, PreTrainedModel

from ...utils import logging


_lowercase			:			int										=						logging.get_logger(__name__)

class 	__SCREAMING_SNAKE_CASE     (				lowerCAmelCase_       ):


					'''simple docstring'''





					_a             =							CLIPConfig

					_a             =							['CLIPEncoderLayer']



					def __init__(	self							:  int,  lowerCamelCase							:  CLIPConfig					)-> List[Any]:
								super().__init__(lowerCamelCase					)

								lowerCamelCase__		:		Dict		     =CLIPVisionModelWithProjection(config.vision_config					)

								lowerCamelCase__		:		Dict		     =nn.Linear(config.vision_config.projection_dim,  1					)
								lowerCamelCase__		:		Union[str, Any]		     =nn.Linear(config.vision_config.projection_dim,  1					)




					@torch.no_grad()
					def 		snake_case	(	self							:  Dict,  lowerCamelCase							:  Optional[Any],  lowerCamelCase							:  List[str],  lowerCamelCase							:  List[Any]=0.5,  lowerCamelCase							:  Optional[Any]=0.5					)-> Optional[int]:
								lowerCamelCase__		:		Dict		     =self.vision_model(lowerCamelCase					)[0]

								lowerCamelCase__		:		List[Any]		     =self.p_head(lowerCamelCase					)
								lowerCamelCase__		:		Optional[Any]		     =nsfw_detected.flatten()
								lowerCamelCase__		:		Any		     =nsfw_detected > p_threshold
								lowerCamelCase__		:		Optional[Any]		     =nsfw_detected.tolist()

								if any(lowerCamelCase					):
											logger.warning(
											    '''Potential NSFW content was detected in one or more images. A black image will be returned instead.'''
											    ''' Try again with a different prompt and/or seed.'''					)

								for idx, nsfw_detected_ in enumerate(lowerCamelCase					):
											if nsfw_detected_:
														lowerCamelCase__		:		Tuple		     =np.zeros(images[idx].shape					)

								lowerCamelCase__		:		str		     =self.w_head(lowerCamelCase					)
								lowerCamelCase__		:		List[str]		     =watermark_detected.flatten()
								lowerCamelCase__		:		Optional[Any]		     =watermark_detected > w_threshold
								lowerCamelCase__		:		int		     =watermark_detected.tolist()

								if any(lowerCamelCase					):
											logger.warning(
											    '''Potential watermarked content was detected in one or more images. A black image will be returned instead.'''
											    ''' Try again with a different prompt and/or seed.'''					)

								for idx, watermark_detected_ in enumerate(lowerCamelCase					):
											if watermark_detected_:
														lowerCamelCase__		:		Optional[int]		     =np.zeros(images[idx].shape					)

								return images, nsfw_detected, watermark_detected




