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


UpperCamelCase_   	=				logging.get_logger(__name__)


# General docstring
UpperCamelCase_   	=				"""MobileNetV1Config"""

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

# Image classification docstring
UpperCamelCase_   	=				"""google/mobilenet_v1_1.0_224"""
UpperCamelCase_   	=				"""tabby, tabby cat"""


UpperCamelCase_   	=				[
    """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 		_UpperCAmelCase (     _lowerCamelCase  :				List[str]     ,					_lowerCamelCase  :				List[Any]     ,					_lowerCamelCase  :				str=None						)			->  str:
  _lowerCAmelCase				:     Optional[int]       =    {}

  if isinstance(_lowerCamelCase     ,					_lowerCamelCase						):
    _lowerCAmelCase				:     Any       =    model.mobilenet_va
  else:
    _lowerCAmelCase				:     List[str]       =    model

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

  for i in range(13						):
    _lowerCAmelCase				:     Tuple       =    i + 1
    _lowerCAmelCase				:     int       =    i * 2

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

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

  if isinstance(_lowerCamelCase     ,					_lowerCamelCase						):
    _lowerCAmelCase				:     str       =    "MobilenetV1/Logits/Conv2d_1c_1x1/"
    _lowerCAmelCase				:     List[Any]       =    model.classifier.weight
    _lowerCAmelCase				:     int       =    model.classifier.bias

  return tf_to_pt_map


def 		_UpperCAmelCase (     _lowerCamelCase  :				Union[str, Any]     ,					_lowerCamelCase  :				str     ,					_lowerCamelCase  :				Tuple						)			->  Dict:
  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				:     Optional[Any]       =    tf.train.list_variables(_lowerCamelCase						)
  _lowerCAmelCase				:     Any       =    {}
  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				:     Tuple       =    array

  # Build TF to PyTorch weights loading map
  _lowerCAmelCase				:     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				:     Tuple       =    tf_weights[name]

    if "depthwise_weights" in name:
      logger.info("""Transposing depthwise"""						)
      _lowerCAmelCase				:     int       =    np.transpose(_lowerCamelCase     ,					(2, 3, 0, 1)						)
    elif "weights" in name:
      logger.info("""Transposing"""						)
      if len(pointer.shape						) == 2:  # copying into linear layer
        _lowerCAmelCase				:     List[Any]       =    array.squeeze().transpose()
      else:
        _lowerCAmelCase				:     List[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				:     Tuple       =    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 		_UpperCAmelCase (     _lowerCamelCase  :				torch.Tensor     ,					_lowerCamelCase  :				nn.Convad						)			->  Dict:
  _lowerCAmelCase				:     Optional[Any]       =    features.shape[-2:]
  _lowerCAmelCase				:     int       =    conv_layer.stride
  _lowerCAmelCase				:     Optional[Any]       =    conv_layer.kernel_size

  if in_height % stride_height == 0:
    _lowerCAmelCase				:     Union[str, Any]       =    max(kernel_height - stride_height     ,					0						)
  else:
    _lowerCAmelCase				:     List[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				:     Optional[int]       =    max(kernel_width - (in_width % stride_width)     ,					0						)

  _lowerCAmelCase				:     Tuple       =    pad_along_width // 2
  _lowerCAmelCase				:     Tuple       =    pad_along_width - pad_left
  _lowerCAmelCase				:     List[Any]       =    pad_along_height // 2
  _lowerCAmelCase				:     List[Any]       =    pad_along_height - pad_top

  _lowerCAmelCase				:     Dict       =    (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      ,      ):
         super().__init__()
         _lowerCAmelCase				:     int       =    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				:     str       =    nn.Convad(
             in_channels=lowercase_      ,      out_channels=lowercase_      ,      kernel_size=lowercase_      ,      stride=lowercase_      ,      padding=lowercase_      ,      groups=lowercase_      ,      bias=lowercase_      ,      padding_mode="""zeros"""      ,      )

         if use_normalization:
           _lowerCAmelCase				:     Dict       =    nn.BatchNormad(
               num_features=lowercase_      ,      eps=config.layer_norm_eps      ,      momentum=0.9997      ,      affine=lowercase_      ,      track_running_stats=lowercase_      ,      )
         else:
           _lowerCAmelCase				:     Optional[int]       =    None

         if use_activation:
           if isinstance(lowercase_      ,      lowercase_     ):
             _lowerCAmelCase				:     str       =    ACTaFN[use_activation]
           elif isinstance(config.hidden_act      ,      lowercase_     ):
             _lowerCAmelCase				:     int       =    ACTaFN[config.hidden_act]
           else:
             _lowerCAmelCase				:     Tuple       =    config.hidden_act
         else:
           _lowerCAmelCase				:     Optional[Any]       =    None







       def 							__UpperCamelCase				(   self      ,      snake_case_     ):
         if self.config.tf_padding:
           _lowerCAmelCase				:     int       =    apply_tf_padding(lowercase_      ,      self.convolution     )
         _lowerCAmelCase				:     List[str]       =    self.convolution(lowercase_     )
         if self.normalization is not None:
           _lowerCAmelCase				:     Any       =    self.normalization(lowercase_     )
         if self.activation is not None:
           _lowerCAmelCase				:     Any       =    self.activation(lowercase_     )
         return features



class      a_     (__A			):
       __lowerCAmelCase			: int			=					MobileNetVaConfig
       __lowerCAmelCase			: List[Any]			=					load_tf_weights_in_mobilenet_va
       __lowerCAmelCase			: Tuple			=					"mobilenet_v1"
       __lowerCAmelCase			: str			=					"pixel_values"
       __lowerCAmelCase			: Dict			=					False




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



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 ([`MobileNetV1Config`]): 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
            [`MobileNetV1ImageProcessor.__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 MobileNetV1 model outputting raw hidden-states without any specific head on top."""							,	__A							,	)
class      a_     (__A			):




       def __init__(   self      ,      snake_case_      ,      snake_case_ = True     ):
         super().__init__(lowercase_     )
         _lowerCAmelCase				:     str       =    config

         _lowerCAmelCase				:     List[Any]       =    3_2
         _lowerCAmelCase				:     Union[str, Any]       =    max(int(depth * config.depth_multiplier     )      ,      config.min_depth     )

         _lowerCAmelCase				:     Optional[int]       =    MobileNetVaConvLayer(
             lowercase_      ,      in_channels=config.num_channels      ,      out_channels=lowercase_      ,      kernel_size=3      ,      stride=2      ,      )

         _lowerCAmelCase				:     Tuple       =    [1, 2, 1, 2, 1, 2, 1, 1, 1, 1, 1, 2, 1]

         _lowerCAmelCase				:     Any       =    nn.ModuleList()
         for i in range(1_3     ):
           _lowerCAmelCase				:     List[Any]       =    out_channels

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

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

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

         _lowerCAmelCase				:     Optional[Any]       =    nn.AdaptiveAvgPoolad((1, 1)     ) if add_pooling_layer else None

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




       def 							__UpperCamelCase				(   self      ,      snake_case_     ):
         raise NotImplementedError







       @add_start_docstrings_to_model_forward(lowercase_     )
       @add_code_sample_docstrings(
           checkpoint=_CHECKPOINT_FOR_DOC      ,      output_type=lowercase_      ,      config_class=_CONFIG_FOR_DOC      ,      modality="""vision"""      ,      expected_output=_EXPECTED_OUTPUT_SHAPE      ,      )
       def 							__UpperCamelCase				(   self      ,      snake_case_ = None      ,      snake_case_ = None      ,      snake_case_ = None      ,      ):
         _lowerCAmelCase				:     str       =    (
             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				:     Dict       =    self.conv_stem(lowercase_     )

         _lowerCAmelCase				:     Union[str, Any]       =    () if output_hidden_states else None

         for i, layer_module in enumerate(self.layer     ):
           _lowerCAmelCase				:     List[Any]       =    layer_module(lowercase_     )

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

         _lowerCAmelCase				:     int       =    hidden_states

         if self.pooler is not None:
           _lowerCAmelCase				:     Any       =    torch.flatten(self.pooler(lowercase_     )      ,      start_dim=1     )
         else:
           _lowerCAmelCase				:     List[str]       =    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=lowercase_      ,      pooler_output=lowercase_      ,      hidden_states=lowercase_      ,      )



@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    """							,	__A							,	)
class      a_     (__A			):




       def __init__(   self      ,      snake_case_     ):
         super().__init__(lowercase_     )

         _lowerCAmelCase				:     int       =    config.num_labels
         _lowerCAmelCase				:     Optional[int]       =    MobileNetVaModel(lowercase_     )

         _lowerCAmelCase				:     Optional[int]       =    self.mobilenet_va.layer[-1].convolution.out_channels

         # Classifier head
         _lowerCAmelCase				:     Optional[Any]       =    nn.Dropout(config.classifier_dropout_prob      ,      inplace=lowercase_     )
         _lowerCAmelCase				:     Dict       =    nn.Linear(lowercase_      ,      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(lowercase_     )
       @add_code_sample_docstrings(
           checkpoint=_IMAGE_CLASS_CHECKPOINT      ,      output_type=lowercase_      ,      config_class=_CONFIG_FOR_DOC      ,      expected_output=_IMAGE_CLASS_EXPECTED_OUTPUT      ,      )
       def 							__UpperCamelCase				(   self      ,      snake_case_ = None      ,      snake_case_ = None      ,      snake_case_ = None      ,      snake_case_ = None      ,      ):
         _lowerCAmelCase				:     Dict       =    return_dict if return_dict is not None else self.config.use_return_dict

         _lowerCAmelCase				:     Any       =    self.mobilenet_va(lowercase_      ,      output_hidden_states=lowercase_      ,      return_dict=lowercase_     )

         _lowerCAmelCase				:     int       =    outputs.pooler_output if return_dict else outputs[1]

         _lowerCAmelCase				:     Any       =    self.classifier(self.dropout(lowercase_     )     )

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

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

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

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



import unittest

from datasets import load_dataset

from transformers.pipelines import pipeline
from transformers.testing_utils import is_pipeline_test, nested_simplify, require_torch, slow



@is_pipeline_test
@require_torch
class   snake_case_       (		unittest.TestCase       ):






   @require_torch
   def       __UpperCamelCase     (      self				:     Optional[int]  )					->						List[Any]:
       lowercase__ :   Union[str, Any]    					=				pipeline(
           task="zero-shot-audio-classification"	,						model="hf-internal-testing/tiny-clap-htsat-unfused"  )
       lowercase__ :   List[str]    					=				load_dataset("ashraq/esc50"  )
       lowercase__ :   List[Any]    					=				dataset["train"]["audio"][-1]["array"]
       lowercase__ :   Dict    					=				audio_classifier(lowercase_	,						candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]  )
       self.assertEqual(
           nested_simplify(lowercase_  )	,						[{"score": 0.5_01, "label": "Sound of a dog"}, {"score": 0.4_99, "label": "Sound of vaccum cleaner"}]	,						)






   @unittest.skip("No models are available in TF"  )
   def       __UpperCamelCase     (      self				:     str  )					->						Optional[int]:
       pass






   @slow
   @require_torch
   def       __UpperCamelCase     (      self				:     List[str]  )					->						int:
       lowercase__ :   Tuple    					=				pipeline(
           task="zero-shot-audio-classification"	,						model="laion/clap-htsat-unfused"	,						)
       # This is an audio of a dog
       lowercase__ :   Union[str, Any]    					=				load_dataset("ashraq/esc50"  )
       lowercase__ :   Tuple    					=				dataset["train"]["audio"][-1]["array"]
       lowercase__ :   List[Any]    					=				audio_classifier(lowercase_	,						candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]  )

       self.assertEqual(
           nested_simplify(lowercase_  )	,						[
               {"score": 0.9_99, "label": "Sound of a dog"},
               {"score": 0.0_01, "label": "Sound of vaccum cleaner"},
           ]	,						)

       lowercase__ :   int    					=				audio_classifier([audio] * 5	,						candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]  )
       self.assertEqual(
           nested_simplify(lowercase_  )	,						[
               [
                   {"score": 0.9_99, "label": "Sound of a dog"},
                   {"score": 0.0_01, "label": "Sound of vaccum cleaner"},
               ],
           ]
           * 5	,						)
       lowercase__ :   Tuple    					=				audio_classifier(
           [audio] * 5	,						candidate_labels=["Sound of a dog", "Sound of vaccum cleaner"]	,						batch_size=5  )
       self.assertEqual(
           nested_simplify(lowercase_  )	,						[
               [
                   {"score": 0.9_99, "label": "Sound of a dog"},
                   {"score": 0.0_01, "label": "Sound of vaccum cleaner"},
               ],
           ]
           * 5	,						)







   @unittest.skip("No models are available in TF"  )
   def       __UpperCamelCase     (      self				:     Union[str, Any]  )					->						Dict:
       pass



