




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


lowerCAmelCase_						=			logging.get_logger(__name__)


# General docstring
lowerCAmelCase_						=			'MobileNetV1Config'

# Base docstring
lowerCAmelCase_						=			'google/mobilenet_v1_1.0_224'
lowerCAmelCase_						=			[1, 10_24, 7, 7]

# Image classification docstring
lowerCAmelCase_						=			'google/mobilenet_v1_1.0_224'
lowerCAmelCase_						=			'tabby, tabby cat'


lowerCAmelCase_						=			[
    '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(    __magic_name__			,		__magic_name__			,		__magic_name__=None   )			->							List[str]:





     '''simple docstring'''


     lowercase :       Any       =			{}

     if isinstance(__magic_name__			,		__magic_name__   ):
          lowercase :       Tuple       =			model.mobilenet_va
     else:
          lowercase :       int       =			model

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

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

          lowercase :       Dict       =			backbone.layer[pt_index]
          lowercase :       List[str]       =			F"""MobilenetV1/Conv2d_{tf_index}_depthwise/"""
          lowercase :       str       =			pointer.convolution.weight
          lowercase :       Optional[int]       =			pointer.normalization.bias
          lowercase :       List[Any]       =			pointer.normalization.weight
          lowercase :       Optional[int]       =			pointer.normalization.running_mean
          lowercase :       Union[str, Any]       =			pointer.normalization.running_var

          lowercase :       Any       =			backbone.layer[pt_index + 1]
          lowercase :       Tuple       =			F"""MobilenetV1/Conv2d_{tf_index}_pointwise/"""
          lowercase :       List[str]       =			pointer.convolution.weight
          lowercase :       str       =			pointer.normalization.bias
          lowercase :       int       =			pointer.normalization.weight
          lowercase :       str       =			pointer.normalization.running_mean
          lowercase :       Any       =			pointer.normalization.running_var

     if isinstance(__magic_name__			,		__magic_name__   ):
          lowercase :       List[str]       =			'''MobilenetV1/Logits/Conv2d_1c_1x1/'''
          lowercase :       Any       =			model.classifier.weight
          lowercase :       str       =			model.classifier.bias

     return tf_to_pt_map






def  snake_case(    __magic_name__			,		__magic_name__			,		__magic_name__   )			->							Any:





     '''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
     lowercase :       List[str]       =			tf.train.list_variables(__magic_name__   )
     lowercase :       Union[str, Any]       =			{}
     for name, shape in init_vars:
          logger.info(F"""Loading TF weight {name} with shape {shape}"""   )
          lowercase :       int       =			tf.train.load_variable(__magic_name__			,		__magic_name__   )
          lowercase :       str       =			array

     # Build TF to PyTorch weights loading map
     lowercase :       Dict       =			_build_tf_to_pytorch_map(__magic_name__			,		__magic_name__			,		__magic_name__   )

     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

          lowercase :       Any       =			tf_weights[name]

          if "depthwise_weights" in name:
               logger.info('''Transposing depthwise'''   )
               lowercase :       str       =			np.transpose(__magic_name__			,		(2, 3, 0, 1)   )
          elif "weights" in name:
               logger.info('''Transposing'''   )
               if len(pointer.shape   ) == 2:  # copying into linear layer
                    lowercase :       Tuple       =			array.squeeze().transpose()
               else:
                    lowercase :       Optional[Any]       =			np.transpose(__magic_name__			,		(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}"""   )
          lowercase :       Union[str, Any]       =			torch.from_numpy(__magic_name__   )

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

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






def  snake_case(    __magic_name__			,		__magic_name__   )			->							torch.Tensor:





     '''simple docstring'''


     lowercase      ,	lowercase :       Dict       =			features.shape[-2:]
     lowercase      ,	lowercase :       List[Any]       =			conv_layer.stride
     lowercase      ,	lowercase :       int       =			conv_layer.kernel_size

     if in_height % stride_height == 0:
          lowercase :       Tuple       =			max(kernel_height - stride_height			,		0   )
     else:
          lowercase :       Any       =			max(kernel_height - (in_height % stride_height)			,		0   )

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

     lowercase :       Union[str, Any]       =			pad_along_width // 2
     lowercase :       str       =			pad_along_width - pad_left
     lowercase :       str       =			pad_along_height // 2
     lowercase :       Union[str, Any]       =			pad_along_height - pad_top

     lowercase :       List[str]       =			(pad_left, pad_right, pad_top, pad_bottom)
     return nn.functional.pad(__magic_name__			,		__magic_name__			,		'''constant'''			,		0.0   )
class 						_A (  nn.Module				):






  def __init__(  self		:     Any				,						_A		:     MobileNetVaConfig				,						_A		:     int				,						_A		:     int				,						_A		:     int				,						_A		:     Optional[int] = 1				,						_A		:     Optional[int] = 1				,						_A		:     bool = False				,						_A		:     Optional[bool] = True				,						_A		:     Optional[bool or str] = True				,						)		->   None:






       """simple docstring"""







       super().__init__()
       lowercase :       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."""				)

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

       lowercase :       int       =			nn.Convad(
           in_channels=_A				,						out_channels=_A				,						kernel_size=_A				,						stride=_A				,						padding=_A				,						groups=_A				,						bias=_A				,						padding_mode='''zeros'''				,						)

       if use_normalization:
            lowercase :       Any       =			nn.BatchNormad(
                num_features=_A				,						eps=config.layer_norm_eps				,						momentum=0.9_997				,						affine=_A				,						track_running_stats=_A				,						)
       else:
            lowercase :       Tuple       =			None

       if use_activation:
            if isinstance(_A				,						_A				):
                 lowercase :       Optional[Any]       =			ACTaFN[use_activation]
            elif isinstance(config.hidden_act				,						_A				):
                 lowercase :       Tuple       =			ACTaFN[config.hidden_act]
            else:
                 lowercase :       Any       =			config.hidden_act
       else:
            lowercase :       Dict       =			None






  def  __a      (  self		:     Tuple				,						_A		:     torch.Tensor				)		->   torch.Tensor:






       """simple docstring"""







       if self.config.tf_padding:
            lowercase :       int       =			apply_tf_padding(_A				,						self.convolution				)
       lowercase :       str       =			self.convolution(_A				)
       if self.normalization is not None:
            lowercase :       str       =			self.normalization(_A				)
       if self.activation is not None:
            lowercase :       str       =			self.activation(_A				)
       return features


class 						_A (  _lowerCamelCase				):
  _UpperCamelCase       :      Dict			=      MobileNetVaConfig
  _UpperCamelCase       :      Tuple			=      load_tf_weights_in_mobilenet_va
  _UpperCamelCase       :      List[Any]			=      '''mobilenet_v1'''
  _UpperCamelCase       :      Optional[Any]			=      '''pixel_values'''
  _UpperCamelCase       :      Dict			=      False






  def  __a      (  self		:     int				,						_A		:     Union[nn.Linear, nn.Convad]				)		->   None:






       """simple docstring"""







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


lowerCAmelCase_						=			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'

lowerCAmelCase_						=			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 						_A (  _lowerCamelCase				):






  def __init__(  self		:     Optional[Any]				,						_A		:     MobileNetVaConfig				,						_A		:     bool = True				)		->   Union[str, Any]:






       """simple docstring"""







       super().__init__(_A				)
       lowercase :       Optional[Any]       =			config

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

       lowercase :       Union[str, Any]       =			MobileNetVaConvLayer(
           _A				,						in_channels=config.num_channels				,						out_channels=_A				,						kernel_size=3				,						stride=2				,						)

       lowercase :       Any       =			[1, 2, 1, 2, 1, 2, 1, 1, 1, 1, 1, 2, 1]

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

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

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

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

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

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






  def  __a      (  self		:     Any				,						_A		:     Optional[Any]				)		->   Any:






       """simple docstring"""







       raise NotImplementedError






  @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		:     str				,						_A		:     Optional[torch.Tensor] = None				,						_A		:     Optional[bool] = None				,						_A		:     Optional[bool] = None				,						)		->   Union[tuple, BaseModelOutputWithPoolingAndNoAttention]:






       """simple docstring"""







       lowercase :       Tuple       =			(
           output_hidden_states if output_hidden_states is not None else self.config.output_hidden_states
       )
       lowercase :       int       =			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'''				)

       lowercase :       Dict       =			self.conv_stem(_A				)

       lowercase :       List[Any]       =			() if output_hidden_states else None

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

            if output_hidden_states:
                 lowercase :       Any       =			all_hidden_states + (hidden_states,)

       lowercase :       Tuple       =			hidden_states

       if self.pooler is not None:
            lowercase :       Any       =			torch.flatten(self.pooler(_A				)				,						start_dim=1				)
       else:
            lowercase :       Optional[int]       =			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=_A				,						pooler_output=_A				,						hidden_states=_A				,						)


@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.
    '''      ,       _lowerCamelCase      ,       )
class 						_A (  _lowerCamelCase				):






  def __init__(  self		:     Optional[Any]				,						_A		:     MobileNetVaConfig				)		->   None:






       """simple docstring"""







       super().__init__(_A				)

       lowercase :       Dict       =			config.num_labels
       lowercase :       List[Any]       =			MobileNetVaModel(_A				)

       lowercase :       Dict       =			self.mobilenet_va.layer[-1].convolution.out_channels

       # Classifier head
       lowercase :       Union[str, Any]       =			nn.Dropout(config.classifier_dropout_prob				,						inplace=_A				)
       lowercase :       List[str]       =			nn.Linear(_A				,						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		:     Dict				,						_A		:     Optional[torch.Tensor] = None				,						_A		:     Optional[bool] = None				,						_A		:     Optional[torch.Tensor] = None				,						_A		:     Optional[bool] = None				,						)		->   Union[tuple, ImageClassifierOutputWithNoAttention]:






       """simple docstring"""







       lowercase :       Dict       =			return_dict if return_dict is not None else self.config.use_return_dict

       lowercase :       str       =			self.mobilenet_va(_A				,						output_hidden_states=_A				,						return_dict=_A				)

       lowercase :       Any       =			outputs.pooler_output if return_dict else outputs[1]

       lowercase :       Tuple       =			self.classifier(self.dropout(_A				)				)

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

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

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

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





class 						_A :  # Public class to implement a graph






  def __init__(  self		:     List[Any]				,						_A		:     int				,						_A		:     int				,						_A		:     list[list[bool]]				)		->   None:






       """simple docstring"""







       lowercase :       Tuple       =			row
       lowercase :       Union[str, Any]       =			col
       lowercase :       int       =			graph






  def  __a      (  self		:     List[Any]				,						_A		:     int				,						_A		:     int				,						_A		:     list[list[bool]]				)		->   bool:






       """simple docstring"""







       return (
           0 <= i < self.ROW
           and 0 <= j < self.COL
           and not visited[i][j]
           and self.graph[i][j]
       )






  def  __a      (  self		:     int				,						_A		:     int				,						_A		:     int				,						_A		:     list[list[bool]]				)		->   None:






       """simple docstring"""







       lowercase :       List[str]       =			[-1, -1, -1, 0, 0, 1, 1, 1]  # Coordinate order
       lowercase :       Dict       =			[-1, 0, 1, -1, 1, -1, 0, 1]
       lowercase :       Dict       =			True  # Make those cells visited
       for k in range(8				):
            if self.is_safe(i + row_nbr[k]				,						j + col_nbr[k]				,						_A				):
                 self.diffs(i + row_nbr[k]				,						j + col_nbr[k]				,						_A				)






  def  __a      (  self		:     List[str]				)		->   int:  # And finally, count all islands.






       """simple docstring"""







       lowercase :       List[str]       =			[[False for j in range(self.COL				)] for i in range(self.ROW				)]
       lowercase :       Optional[Any]       =			0
       for i in range(self.ROW				):
            for j in range(self.COL				):
                 if visited[i][j] is False and self.graph[i][j] == 1:
                      self.diffs(_A				,						_A				,						_A				)
                      count += 1
       return count