
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


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


# General docstring
A     :	Optional[int]    = 'MobileNetV1Config'

# Base docstring
A     :	Any    = 'google/mobilenet_v1_1.0_224'
A     :	List[str]    = [1, 1_0_2_4, 7, 7]

# Image classification docstring
A     :	List[str]    = 'google/mobilenet_v1_1.0_224'
A     :	List[str]    = 'tabby, tabby cat'


A     :	Tuple    = [
    '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     __lowerCAmelCase			(			a__						,    a__						,    a__=None  )       ->  List[Any]:
				__a										=    {}

				if isinstance(lowerCamelCase_						,    lowerCamelCase_  ):
								__a										=    model.mobilenet_va
				else:
								__a										=    model

				__a										=    '''MobilenetV1/Conv2d_0/'''
				__a										=    backbone.conv_stem.convolution.weight
				__a										=    backbone.conv_stem.normalization.bias
				__a										=    backbone.conv_stem.normalization.weight
				__a										=    backbone.conv_stem.normalization.running_mean
				__a										=    backbone.conv_stem.normalization.running_var

				for i in range(13  ):
								__a										=    i + 1
								__a										=    i * 2

								__a										=    backbone.layer[pt_index]
								__a										=    F"""MobilenetV1/Conv2d_{tf_index}_depthwise/"""
								__a										=    pointer.convolution.weight
								__a										=    pointer.normalization.bias
								__a										=    pointer.normalization.weight
								__a										=    pointer.normalization.running_mean
								__a										=    pointer.normalization.running_var

								__a										=    backbone.layer[pt_index + 1]
								__a										=    F"""MobilenetV1/Conv2d_{tf_index}_pointwise/"""
								__a										=    pointer.convolution.weight
								__a										=    pointer.normalization.bias
								__a										=    pointer.normalization.weight
								__a										=    pointer.normalization.running_mean
								__a										=    pointer.normalization.running_var

				if isinstance(lowerCamelCase_						,    lowerCamelCase_  ):
								__a										=    '''MobilenetV1/Logits/Conv2d_1c_1x1/'''
								__a										=    model.classifier.weight
								__a										=    model.classifier.bias

				return tf_to_pt_map


def     __lowerCAmelCase			(			a__						,    a__						,    a__  )       ->  Union[str, Any]:
				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
				__a										=    tf.train.list_variables(lowerCamelCase_  )
				__a										=    {}
				for name, shape in init_vars:
								logger.info(F"""Loading TF weight {name} with shape {shape}"""  )
								__a										=    tf.train.load_variable(lowerCamelCase_						,    lowerCamelCase_  )
								__a										=    array

				# Build TF to PyTorch weights loading map
				__a										=    _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

								__a										=    tf_weights[name]

								if "depthwise_weights" in name:
												logger.info('''Transposing depthwise'''  )
												__a										=    np.transpose(lowerCamelCase_						,    (2, 3, 0, 1)  )
								elif "weights" in name:
												logger.info('''Transposing'''  )
												if len(pointer.shape  ) == 2:  # copying into linear layer
																__a										=    array.squeeze().transpose()
												else:
																__a										=    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}"""  )
								__a										=    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     __lowerCAmelCase			(			a__						,    a__  )       ->  Union[str, Any]:
				__a							,     __a										=    features.shape[-2:]
				__a							,     __a										=    conv_layer.stride
				__a							,     __a										=    conv_layer.kernel_size

				if in_height % stride_height == 0:
								__a										=    max(kernel_height - stride_height						,    0  )
				else:
								__a										=    max(kernel_height - (in_height % stride_height)						,    0  )

				if in_width % stride_width == 0:
								__a										=    max(kernel_width - stride_width						,    0  )
				else:
								__a										=    max(kernel_width - (in_width % stride_width)						,    0  )

				__a										=    pad_along_width // 2
				__a										=    pad_along_width - pad_left
				__a										=    pad_along_height // 2
				__a										=    pad_along_height - pad_top

				__a										=    (pad_left, pad_right, pad_top, pad_bottom)
				return nn.functional.pad(lowerCamelCase_						,    lowerCamelCase_						,    '''constant'''						,    0.0  )







class 			__A( nn.Module			):




				def __init__(     self		,     _snake_case		,     _snake_case		,     _snake_case		,     _snake_case		,     _snake_case = 1		,     _snake_case = 1		,     _snake_case = False		,     _snake_case = True		,     _snake_case = True		,     )    ->      Any:


								'''simple docstring'''




								super().__init__()
								__a										=    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."""				)

								__a										=    0 if config.tf_padding else int((kernel_size - 1) / 2				)

								__a										=    nn.Convad(
								    in_channels=_snake_case		,     out_channels=_snake_case		,     kernel_size=_snake_case		,     stride=_snake_case		,     padding=_snake_case		,     groups=_snake_case		,     bias=_snake_case		,     padding_mode='''zeros'''		,     )

								if use_normalization:
												__a										=    nn.BatchNormad(
												    num_features=_snake_case		,     eps=config.layer_norm_eps		,     momentum=0.9997		,     affine=_snake_case		,     track_running_stats=_snake_case		,     )
								else:
												__a										=    None

								if use_activation:
												if isinstance(_snake_case		,     _snake_case				):
																__a										=    ACTaFN[use_activation]
												elif isinstance(config.hidden_act		,     _snake_case				):
																__a										=    ACTaFN[config.hidden_act]
												else:
																__a										=    config.hidden_act
								else:
												__a										=    None




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case				)    ->      Tuple:


								'''simple docstring'''




								if self.config.tf_padding:
												__a										=    apply_tf_padding(_snake_case		,     self.convolution				)
								__a										=    self.convolution(_snake_case				)
								if self.normalization is not None:
												__a										=    self.normalization(_snake_case				)
								if self.activation is not None:
												__a										=    self.activation(_snake_case				)
								return features







class 			__A( A__			):
				snake_case_						   =    MobileNetVaConfig
				snake_case_						   =    load_tf_weights_in_mobilenet_va
				snake_case_						   =    '''mobilenet_v1'''
				snake_case_						   =    '''pixel_values'''
				snake_case_						   =    False




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case				)    ->      Any:


								'''simple docstring'''




								if isinstance(_snake_case		,     (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(_snake_case		,     nn.BatchNormad				):
												module.bias.data.zero_()
												module.weight.data.fill_(1.0				)



A     :	Optional[Any]    = 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'

A     :	Optional[Any]    = 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.'''  ,   A__  ,   )
class 			__A( A__			):




				def __init__(     self		,     _snake_case		,     _snake_case = True				)    ->      List[str]:


								'''simple docstring'''




								super().__init__(_snake_case				)
								__a										=    config

								__a										=    32
								__a										=    max(int(depth * config.depth_multiplier				)		,     config.min_depth				)

								__a										=    MobileNetVaConvLayer(
								    _snake_case		,     in_channels=config.num_channels		,     out_channels=_snake_case		,     kernel_size=3		,     stride=2		,     )

								__a										=    [1, 2, 1, 2, 1, 2, 1, 1, 1, 1, 1, 2, 1]

								__a										=    nn.ModuleList()
								for i in range(13				):
												__a										=    out_channels

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

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

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

								__a										=    nn.AdaptiveAvgPoolad((1, 1)				) if add_pooling_layer else None

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




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case				)    ->      Any:


								'''simple docstring'''




								raise NotImplementedError




				@add_start_docstrings_to_model_forward(_snake_case				)
				@add_code_sample_docstrings(
				    checkpoint=_CHECKPOINT_FOR_DOC		,     output_type=_snake_case		,     config_class=_CONFIG_FOR_DOC		,     modality='''vision'''		,     expected_output=_EXPECTED_OUTPUT_SHAPE		,     )
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case = None		,     _snake_case = None		,     _snake_case = None		,     )    ->      str:


								'''simple docstring'''




								__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

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

								__a										=    self.conv_stem(_snake_case				)

								__a										=    () if output_hidden_states else None

								for i, layer_module in enumerate(self.layer				):
												__a										=    layer_module(_snake_case				)

												if output_hidden_states:
																__a										=    all_hidden_states + (hidden_states,)

								__a										=    hidden_states

								if self.pooler is not None:
												__a										=    torch.flatten(self.pooler(_snake_case				)		,     start_dim=1				)
								else:
												__a										=    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=_snake_case		,     pooler_output=_snake_case		,     hidden_states=_snake_case		,     )







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




				def __init__(     self		,     _snake_case				)    ->      str:


								'''simple docstring'''




								super().__init__(_snake_case				)

								__a										=    config.num_labels
								__a										=    MobileNetVaModel(_snake_case				)

								__a										=    self.mobilenet_va.layer[-1].convolution.out_channels

								# Classifier head
								__a										=    nn.Dropout(config.classifier_dropout_prob		,     inplace=_snake_case				)
								__a										=    nn.Linear(_snake_case		,     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(_snake_case				)
				@add_code_sample_docstrings(
				    checkpoint=_IMAGE_CLASS_CHECKPOINT		,     output_type=_snake_case		,     config_class=_CONFIG_FOR_DOC		,     expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT		,     )
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case = None		,     _snake_case = None		,     _snake_case = None		,     _snake_case = None		,     )    ->      Optional[Any]:


								'''simple docstring'''




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

								__a										=    self.mobilenet_va(_snake_case		,     output_hidden_states=_snake_case		,     return_dict=_snake_case				)

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

								__a										=    self.classifier(self.dropout(_snake_case				)				)

								__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(_snake_case		,     _snake_case				)
												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(_snake_case		,     _snake_case				)

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

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







import argparse
import os

import numpy as np
import tensorflow as tf
import torch

from transformers import BertModel






def   a			(						lowerCamelCase_		,   lowerCamelCase_		,   lowerCamelCase_				):






				'''simple docstring'''
				lowercase__    =			('''dense.weight''', '''attention.self.query''', '''attention.self.key''', '''attention.self.value''')

				lowercase__    =			(
				    ('''layer.''', '''layer_'''),
				    ('''word_embeddings.weight''', '''word_embeddings'''),
				    ('''position_embeddings.weight''', '''position_embeddings'''),
				    ('''token_type_embeddings.weight''', '''token_type_embeddings'''),
				    ('''.''', '''/'''),
				    ('''LayerNorm/weight''', '''LayerNorm/gamma'''),
				    ('''LayerNorm/bias''', '''LayerNorm/beta'''),
				    ('''weight''', '''kernel'''),
				)

				if not os.path.isdir(lowerCamelCase_				):
								os.makedirs(lowerCamelCase_				)

				lowercase__    =			model.state_dict()

				def to_tf_var_name(lowerCamelCase_				):
								for patt, repl in iter(lowerCamelCase_				):
												lowercase__    =			name.replace(lowerCamelCase_		,   lowerCamelCase_				)
								return F"""bert/{name}"""

				def create_tf_var(lowerCamelCase_		,   lowerCamelCase_		,   lowerCamelCase_				):
								lowercase__    =			tf.dtypes.as_dtype(tensor.dtype				)
								lowercase__    =			tf.get_variable(dtype=lowerCamelCase_		,   shape=tensor.shape		,   name=lowerCamelCase_		,   initializer=tf.zeros_initializer()				)
								session.run(tf.variables_initializer([tf_var]				)				)
								session.run(lowerCamelCase_				)
								return tf_var

				tf.reset_default_graph()
				with tf.Session() as session:
								for var_name in state_dict:
												lowercase__    =			to_tf_var_name(lowerCamelCase_				)
												lowercase__    =			state_dict[var_name].numpy()
												if any(x in var_name for x in tensors_to_transpose				):
																lowercase__    =			torch_tensor.T
												lowercase__    =			create_tf_var(tensor=lowerCamelCase_		,   name=lowerCamelCase_		,   session=lowerCamelCase_				)
												tf.keras.backend.set_value(lowerCamelCase_		,   lowerCamelCase_				)
												lowercase__    =			session.run(lowerCamelCase_				)
												print(F"""Successfully created {tf_name}: {np.allclose(lowerCamelCase_		,   lowerCamelCase_				)}"""				)

								lowercase__    =			tf.train.Saver(tf.trainable_variables()				)
								saver.save(lowerCamelCase_		,   os.path.join(lowerCamelCase_		,   model_name.replace('''-'''		,   '''_'''				) + '''.ckpt'''				)				)






def   a			(						lowerCamelCase_=None				):






				'''simple docstring'''
				lowercase__    =			argparse.ArgumentParser()
				parser.add_argument('''--model_name'''		,   type=lowerCamelCase_		,   required=lowerCamelCase_		,   help='''model name e.g. bert-base-uncased'''				)
				parser.add_argument(
				    '''--cache_dir'''		,   type=lowerCamelCase_		,   default=lowerCamelCase_		,   required=lowerCamelCase_		,   help='''Directory containing pytorch model'''				)
				parser.add_argument('''--pytorch_model_path'''		,   type=lowerCamelCase_		,   required=lowerCamelCase_		,   help='''/path/to/<pytorch-model-name>.bin'''				)
				parser.add_argument('''--tf_cache_dir'''		,   type=lowerCamelCase_		,   required=lowerCamelCase_		,   help='''Directory in which to save tensorflow model'''				)
				lowercase__    =			parser.parse_args(lowerCamelCase_				)

				lowercase__    =			BertModel.from_pretrained(
				    pretrained_model_name_or_path=args.model_name		,   state_dict=torch.load(args.pytorch_model_path				)		,   cache_dir=args.cache_dir		,   )

				convert_pytorch_checkpoint_to_tf(model=lowerCamelCase_		,   ckpt_dir=args.tf_cache_dir		,   model_name=args.model_name				)


if __name__ == "__main__":
							main()




