





'''simple docstring'''




import argparse
import logging
import os
from datetime import datetime

import numpy as np
import torch
from torch import nn
from torch.utils.data import DataLoader, RandomSampler, TensorDataset
from tqdm import tqdm

from transformers import GPTaLMHeadModel


lowercase__       : List[str]         =       logging.getLogger(__name__)






def 			_lowerCAmelCase     (    __snake_case    :					List[str]					,				__snake_case    :					List[Any]     )	->	int:
     # save results
     if os.path.exists(__snake_case     ):
          if os.path.exists(os.path.join(__snake_case					,				'config.json'     )     ) and os.path.isfile(
              os.path.join(__snake_case					,				'config.json'     )     ):
               os.remove(os.path.join(__snake_case					,				'config.json'     )     )
          if os.path.exists(os.path.join(__snake_case					,				'pytorch_model.bin'     )     ) and os.path.isfile(
              os.path.join(__snake_case					,				'pytorch_model.bin'     )     ):
               os.remove(os.path.join(__snake_case					,				'pytorch_model.bin'     )     )
     else:
          os.makedirs(__snake_case     )
     model.save_pretrained(__snake_case     )






def 			_lowerCAmelCase     (    __snake_case    :					Dict					,				__snake_case    :					List[Any]=False     )	->	Any:
     __A		:					Optional[Any]						=			2
     if unlogit:
          __A		:					Optional[Any]						=			torch.pow(__snake_case					,				__snake_case     )
     __A		:					Tuple						=			p * torch.log(__snake_case     )
     __A		:					Any						=			0
     return -plogp.sum(dim=-1     )






def 			_lowerCAmelCase     (    __snake_case    :					Optional[Any]     )	->	Optional[int]:
     logger.info('lv, h >\t' + '\t'.join(f'{x + 1}' for x in range(len(__snake_case     )     )     )     )
     for row in range(len(__snake_case     )     ):
          if tensor.dtype != torch.long:
               logger.info(f'layer {row + 1}:\t' + '\t'.join(f'{x:.5f}' for x in tensor[row].cpu().data     )     )
          else:
               logger.info(f'layer {row + 1}:\t' + '\t'.join(f'{x:d}' for x in tensor[row].cpu().data     )     )






def 			_lowerCAmelCase     (    __snake_case    :					Any					,				__snake_case    :					Optional[Any]					,				__snake_case    :					str					,				__snake_case    :					List[Any]=True					,				__snake_case    :					str=True					,				__snake_case    :					List[Any]=None					,				__snake_case    :					Union[str, Any]=False     )	->	Optional[Any]:
     __A					,__A		:					str						=			model.config.num_hidden_layers, model.config.num_attention_heads
     __A		:					Dict						=			torch.zeros(__snake_case					,				__snake_case     ).to(args.device     )
     __A		:					str						=			torch.zeros(__snake_case					,				__snake_case     ).to(args.device     )

     if head_mask is None:
          __A		:					List[Any]						=			torch.ones(__snake_case					,				__snake_case     ).to(args.device     )

     head_mask.requires_grad_(requires_grad=__snake_case     )
     # If actually pruned attention multi-head, set head mask to None to avoid shape mismatch
     if actually_pruned:
          __A		:					int						=			None

     __A		:					int						=			0.0
     __A		:					Optional[Any]						=			0.0
     for step, inputs in enumerate(tqdm(__snake_case					,				desc='Iteration'					,				disable=args.local_rank not in [-1, 0]     )     ):
          __A		:					Dict						=			tuple(t.to(args.device     ) for t in inputs     )
          ((__A)					,)		:					Union[str, Any]						=			inputs

          # Do a forward pass (not with torch.no_grad() since we need gradients for importance score - see below)
          __A		:					Any						=			model(__snake_case					,				labels=__snake_case					,				head_mask=__snake_case     )
          #  (loss), lm_logits, presents, (all hidden_states), (attentions)
          __A					,__A					,__A		:					Union[str, Any]						=			(
              outputs[0],
              outputs[1],
              outputs[-1],
          )  # Loss and logits are the first, attention the last
          loss.backward()  # Backpropagate to populate the gradients in the head mask
          total_loss += loss.detach().cpu().numpy()
          if compute_entropy:
               for layer, attn in enumerate(__snake_case     ):
                    __A		:					List[Any]						=			entropy(attn.detach()					,				__snake_case     )
                    attn_entropy[layer] += masked_entropy.sum(-1     ).sum(0     ).sum(0     ).detach()

          if compute_importance:
               head_importance += head_mask.grad.abs().detach()
          tot_tokens += torch.ones_like(__snake_case     ).float().detach().sum().data

     # Normalize
     attn_entropy /= tot_tokens
     head_importance /= tot_tokens
     # Layerwise importance normalization
     if not args.dont_normalize_importance_by_layer:
          __A		:					Tuple						=			2
          __A		:					Dict						=			torch.pow(torch.pow(__snake_case					,				__snake_case     ).sum(-1     )					,				1 / exponent     )
          head_importance /= norm_by_layer.unsqueeze(-1     ) + 1e-2_0

     if not args.dont_normalize_global_importance:
          __A		:					Any						=			(head_importance - head_importance.min()) / (head_importance.max() - head_importance.min())

     # Print matrices
     if compute_entropy:
          logger.info('Attention entropies'     )
          print_ad_tensor(__snake_case     )
     if compute_importance:
          logger.info('Head importance scores'     )
          print_ad_tensor(__snake_case     )
     logger.info('Head ranked by importance scores'     )
     __A		:					Any						=			torch.zeros(head_importance.numel()					,				dtype=torch.long					,				device=args.device     )
     __A		:					int						=			torch.arange(
         head_importance.numel()					,				device=args.device     )
     __A		:					Tuple						=			head_ranks.view_as(__snake_case     )
     print_ad_tensor(__snake_case     )
     return attn_entropy, head_importance, total_loss






def 			_lowerCAmelCase     (    __snake_case    :					Any					,				__snake_case    :					Optional[Any]					,				__snake_case    :					Any     )	->	Tuple:
     __A					,__A					,__A		:					List[Any]						=			compute_heads_importance(__snake_case					,				__snake_case					,				__snake_case					,				compute_entropy=__snake_case     )
     __A		:					Union[str, Any]						=			1 / loss  # instead of downsteam score use the LM loss
     logger.info('Pruning: original score: %f, threshold: %f'					,				__snake_case					,				original_score * args.masking_threshold     )

     __A		:					Optional[int]						=			torch.ones_like(__snake_case     )
     __A		:					Union[str, Any]						=			max(1					,				int(new_head_mask.numel() * args.masking_amount     )     )

     __A		:					List[str]						=			original_score
     while current_score >= original_score * args.masking_threshold:
          __A		:					Any						=			new_head_mask.clone().detach()  # save current head mask
          # heads from least important to most - keep only not-masked heads
          __A		:					Dict						=			float('Inf'     )
          __A		:					List[str]						=			head_importance.view(-1     ).sort()[1]

          if len(__snake_case     ) <= num_to_mask:
               print('BREAK BY num_to_mask'     )
               break

          # mask heads
          __A		:					List[Any]						=			current_heads_to_mask[:num_to_mask]
          logger.info('Heads to mask: %s'					,				str(current_heads_to_mask.tolist()     )     )
          __A		:					Optional[Any]						=			new_head_mask.view(-1     )
          __A		:					Dict						=			0.0
          __A		:					Union[str, Any]						=			new_head_mask.view_as(__snake_case     )
          __A		:					Optional[Any]						=			new_head_mask.clone().detach()
          print_ad_tensor(__snake_case     )

          # Compute metric and head importance again
          __A					,__A					,__A		:					Tuple						=			compute_heads_importance(
              __snake_case					,				__snake_case					,				__snake_case					,				compute_entropy=__snake_case					,				head_mask=__snake_case     )
          __A		:					Tuple						=			1 / loss
          logger.info(
              'Masking: current score: %f, remaining heads %d (%.1f percents)'					,				__snake_case					,				new_head_mask.sum()					,				new_head_mask.sum() / new_head_mask.numel() * 1_00					,				)

     logger.info('Final head mask'     )
     print_ad_tensor(__snake_case     )
     np.save(os.path.join(args.output_dir					,				'head_mask.npy'     )					,				head_mask.detach().cpu().numpy()     )

     return head_mask






def 			_lowerCAmelCase     (    __snake_case    :					Optional[Any]					,				__snake_case    :					Union[str, Any]					,				__snake_case    :					str					,				__snake_case    :					int     )	->	int:
     __A		:					Any						=			datetime.now()
     __A					,__A					,__A		:					Union[str, Any]						=			compute_heads_importance(
         __snake_case					,				__snake_case					,				__snake_case					,				compute_entropy=__snake_case					,				compute_importance=__snake_case					,				head_mask=__snake_case     )
     __A		:					Tuple						=			1 / loss
     __A		:					str						=			datetime.now() - before_time

     __A		:					Union[str, Any]						=			sum(p.numel() for p in model.parameters()     )
     __A		:					str						=			{
         layer: (1 - head_mask[layer].long()).nonzero().squeeze().tolist() for layer in range(len(__snake_case     )     )
     }

     for k, v in heads_to_prune.items():
          if isinstance(__snake_case					,				__snake_case     ):
               __A		:					Optional[int]						=			[
                   v,
               ]

     assert sum(len(__snake_case     ) for h in heads_to_prune.values()     ) == (1 - head_mask.long()).sum().item()
     model.prune_heads(__snake_case     )
     __A		:					int						=			sum(p.numel() for p in model.parameters()     )

     __A		:					Dict						=			datetime.now()
     __A					,__A					,__A		:					Any						=			compute_heads_importance(
         __snake_case					,				__snake_case					,				__snake_case					,				compute_entropy=__snake_case					,				compute_importance=__snake_case					,				head_mask=__snake_case					,				actually_pruned=__snake_case					,				)

     __A		:					Dict						=			1 / loss
     __A		:					Optional[Any]						=			datetime.now() - before_time

     logger.info(
         'Pruning: original num of params: %.2e, after pruning %.2e (%.1f percents)'					,				__snake_case					,				__snake_case					,				pruned_num_params / original_num_params * 1_00					,				)
     logger.info('Pruning: score with masking: %f score with pruning: %f'					,				__snake_case					,				__snake_case     )
     logger.info('Pruning: speed ratio (original timing / new timing): %f percents'					,				original_time / new_time * 1_00     )
     save_model(__snake_case					,				args.output_dir     )






def 			_lowerCAmelCase     (    )	->	Optional[int]:
     __A		:					List[str]						=			argparse.ArgumentParser()
     # Required parameters
     parser.add_argument(
         '--data_dir'					,				default=__snake_case					,				type=__snake_case					,				required=__snake_case					,				help='The input data dir. Should contain the .tsv files (or other data files) for the task.'					,				)
     parser.add_argument(
         '--model_name_or_path'					,				default=__snake_case					,				type=__snake_case					,				required=__snake_case					,				help='Path to pretrained model or model identifier from huggingface.co/models'					,				)
     parser.add_argument(
         '--output_dir'					,				default=__snake_case					,				type=__snake_case					,				required=__snake_case					,				help='The output directory where the model predictions and checkpoints will be written.'					,				)

     # Other parameters
     parser.add_argument(
         '--config_name'					,				default=''					,				type=__snake_case					,				help='Pretrained config name or path if not the same as model_name_or_path'					,				)
     parser.add_argument(
         '--tokenizer_name'					,				default=''					,				type=__snake_case					,				help='Pretrained tokenizer name or path if not the same as model_name_or_path'					,				)
     parser.add_argument(
         '--cache_dir'					,				default=__snake_case					,				type=__snake_case					,				help='Where do you want to store the pre-trained models downloaded from s3'					,				)
     parser.add_argument(
         '--data_subset'					,				type=__snake_case					,				default=-1					,				help='If > 0: limit the data to a subset of data_subset instances.'     )
     parser.add_argument(
         '--overwrite_output_dir'					,				action='store_true'					,				help='Whether to overwrite data in output directory'     )
     parser.add_argument(
         '--overwrite_cache'					,				action='store_true'					,				help='Overwrite the cached training and evaluation sets'     )

     parser.add_argument(
         '--dont_normalize_importance_by_layer'					,				action='store_true'					,				help='Don\'t normalize importance score by layers'     )
     parser.add_argument(
         '--dont_normalize_global_importance'					,				action='store_true'					,				help='Don\'t normalize all importance scores between 0 and 1'					,				)

     parser.add_argument(
         '--try_masking'					,				action='store_true'					,				help='Whether to try to mask head until a threshold of accuracy.'     )
     parser.add_argument(
         '--masking_threshold'					,				default=0.9					,				type=__snake_case					,				help='masking threshold in term of metrics (stop masking when metric < threshold * original metric value).'					,				)
     parser.add_argument(
         '--masking_amount'					,				default=0.1					,				type=__snake_case					,				help='Amount to heads to masking at each masking step.'     )
     parser.add_argument('--metric_name'					,				default='acc'					,				type=__snake_case					,				help='Metric to use for head masking.'     )

     parser.add_argument(
         '--max_seq_length'					,				default=1_28					,				type=__snake_case					,				help=(
             'The maximum total input sequence length after WordPiece tokenization. \n'
             'Sequences longer than this will be truncated, sequences shorter padded.'
         )					,				)
     parser.add_argument('--batch_size'					,				default=1					,				type=__snake_case					,				help='Batch size.'     )

     parser.add_argument('--seed'					,				type=__snake_case					,				default=42     )
     parser.add_argument('--local_rank'					,				type=__snake_case					,				default=-1					,				help='local_rank for distributed training on gpus'     )
     parser.add_argument('--no_cuda'					,				action='store_true'					,				help='Whether not to use CUDA when available'     )
     parser.add_argument('--server_ip'					,				type=__snake_case					,				default=''					,				help='Can be used for distant debugging.'     )
     parser.add_argument('--server_port'					,				type=__snake_case					,				default=''					,				help='Can be used for distant debugging.'     )
     __A		:					Tuple						=			parser.parse_args()

     if args.server_ip and args.server_port:
          # Distant debugging - see https://code.visualstudio.com/docs/python/debugging#_attach-to-a-local-script
          import ptvsd

          print('Waiting for debugger attach'     )
          ptvsd.enable_attach(address=(args.server_ip, args.server_port)					,				redirect_output=__snake_case     )
          ptvsd.wait_for_attach()

     # Setup devices and distributed training
     if args.local_rank == -1 or args.no_cuda:
          __A		:					Optional[int]						=			torch.device('cuda' if torch.cuda.is_available() and not args.no_cuda else 'cpu'     )
          __A		:					Optional[Any]						=			0 if args.no_cuda else torch.cuda.device_count()
     else:
          torch.cuda.set_device(args.local_rank     )
          __A		:					Dict						=			torch.device('cuda'					,				args.local_rank     )
          __A		:					int						=			1
          torch.distributed.init_process_group(backend='nccl'     )  # Initializes the distributed backend

     # Setup logging
     logging.basicConfig(level=logging.INFO if args.local_rank in [-1, 0] else logging.WARN     )
     logger.info('device: {} n_gpu: {}, distributed: {}'.format(args.device					,				args.n_gpu					,				bool(args.local_rank != -1     )     )     )

     __A		:					Optional[Any]						=			GPTaLMHeadModel.from_pretrained(args.model_name_or_path     )

     # Distributed and parallel training
     model.to(args.device     )
     if args.local_rank != -1:
          __A		:					List[str]						=			nn.parallel.DistributedDataParallel(
              __snake_case					,				device_ids=[args.local_rank]					,				output_device=args.local_rank					,				find_unused_parameters=__snake_case     )
     elif args.n_gpu > 1:
          __A		:					Tuple						=			nn.DataParallel(__snake_case     )

     # Print/save training arguments
     os.makedirs(args.output_dir					,				exist_ok=__snake_case     )
     torch.save(__snake_case					,				os.path.join(args.output_dir					,				'run_args.bin'     )     )
     logger.info('Training/evaluation parameters %s'					,				__snake_case     )

     # Prepare dataset
     __A		:					Any						=			np.concatenate(
         [
             np.loadtxt(args.data_dir					,				dtype=np.intaa     ),
         ]     )
     __A		:					List[Any]						=			(torch.from_numpy(__snake_case     ),)
     __A		:					List[str]						=			TensorDataset(*__snake_case     )
     __A		:					Any						=			RandomSampler(__snake_case     )
     __A		:					List[str]						=			DataLoader(__snake_case					,				sampler=__snake_case					,				batch_size=args.batch_size     )

     # Compute head entropy and importance score
     compute_heads_importance(__snake_case					,				__snake_case					,				__snake_case     )

     # Try head masking (set heads to zero until the score goes under a threshole)
     # and head pruning (remove masked heads and see the effect on the network)
     if args.try_masking and args.masking_threshold > 0.0 and args.masking_threshold < 1.0:
          __A		:					Optional[int]						=			mask_heads(__snake_case					,				__snake_case					,				__snake_case     )
          prune_heads(__snake_case					,				__snake_case					,				__snake_case					,				__snake_case     )


if __name__ == "__main__":
  main()

import inspect
import unittest

from transformers import MobileNetVaConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
						import torch

						from transformers import MobileNetVaForImageClassification, MobileNetVaForSemanticSegmentation, MobileNetVaModel
						from transformers.models.mobilenet_va.modeling_mobilenet_va import MOBILENET_V2_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
						from PIL import Image

						from transformers import MobileNetVaImageProcessor





class   UpperCamelCase_      (      __a		):







						'''simple docstring'''



						def        SCREAMING_SNAKE_CASE(			self							:Tuple     )  ->Optional[int]:
										lowercase         =      self.config_class(**self.inputs_dict     )
										self.parent.assertTrue(hasattr(lowerCAmelCase__		,    "tf_padding"     )     )
										self.parent.assertTrue(hasattr(lowerCAmelCase__		,    "depth_multiplier"     )     )





class   UpperCamelCase_      :







						'''simple docstring'''



						def __init__(			self							:Dict		,    lowerCAmelCase__							:Any		,    lowerCAmelCase__							:Union[str, Any]=13		,    lowerCAmelCase__							:Tuple=3		,    lowerCAmelCase__							:Optional[int]=32		,    lowerCAmelCase__							:Any=0.25		,    lowerCAmelCase__							:Dict=8		,    lowerCAmelCase__							:Optional[int]=8		,    lowerCAmelCase__							:List[str]=6		,    lowerCAmelCase__							:List[Any]=32		,    lowerCAmelCase__							:Optional[Any]=True		,    lowerCAmelCase__							:str=True		,    lowerCAmelCase__							:Optional[int]=True		,    lowerCAmelCase__							:Dict="relu6"		,    lowerCAmelCase__							:Tuple=1280		,    lowerCAmelCase__							:Dict=0.1		,    lowerCAmelCase__							:List[str]=0.02		,    lowerCAmelCase__							:Dict=True		,    lowerCAmelCase__							:List[Any]=True		,    lowerCAmelCase__							:Tuple=10		,    lowerCAmelCase__							:int=None		,    )  ->List[str]:
										lowercase         =      parent
										lowercase         =      batch_size
										lowercase         =      num_channels
										lowercase         =      image_size
										lowercase         =      depth_multiplier
										lowercase         =      depth_divisible_by
										lowercase         =      min_depth
										lowercase         =      expand_ratio
										lowercase         =      tf_padding
										lowercase         =      output_stride
										lowercase         =      first_layer_is_expansion
										lowercase         =      finegrained_output
										lowercase         =      hidden_act
										lowercase         =      last_hidden_size if finegrained_output else int(last_hidden_size * depth_multiplier     )
										lowercase         =      classifier_dropout_prob
										lowercase         =      use_labels
										lowercase         =      is_training
										lowercase         =      num_labels
										lowercase         =      initializer_range
										lowercase         =      scope





						def        SCREAMING_SNAKE_CASE(			self							:str     )  ->Dict:
										lowercase         =      floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]     )

										lowercase         =      None
										lowercase         =      None
										if self.use_labels:
														lowercase         =      ids_tensor([self.batch_size]		,    self.num_labels     )
														lowercase         =      ids_tensor([self.batch_size, self.image_size, self.image_size]		,    self.num_labels     )

										lowercase         =      self.get_config()

										return config, pixel_values, labels, pixel_labels





						def        SCREAMING_SNAKE_CASE(			self							:Optional[int]     )  ->List[Any]:
										return MobileNetVaConfig(
										    num_channels=self.num_channels		,    image_size=self.image_size		,    depth_multiplier=self.depth_multiplier		,    depth_divisible_by=self.depth_divisible_by		,    min_depth=self.min_depth		,    expand_ratio=self.expand_ratio		,    output_stride=self.output_stride		,    first_layer_is_expansion=self.first_layer_is_expansion		,    finegrained_output=self.finegrained_output		,    hidden_act=self.hidden_act		,    tf_padding=self.tf_padding		,    classifier_dropout_prob=self.classifier_dropout_prob		,    initializer_range=self.initializer_range		,    )





						def        SCREAMING_SNAKE_CASE(			self							:Dict		,    lowerCAmelCase__							:Dict		,    lowerCAmelCase__							:Optional[Any]		,    lowerCAmelCase__							:Optional[Any]		,    lowerCAmelCase__							:Tuple     )  ->Any:
										lowercase         =      MobileNetVaModel(config=lowerCAmelCase__     )
										model.to(lowerCAmelCase__     )
										model.eval()
										lowercase         =      model(lowerCAmelCase__     )
										self.parent.assertEqual(
										    result.last_hidden_state.shape		,    (
										        self.batch_size,
										        self.last_hidden_size,
										        self.image_size // self.output_stride,
										        self.image_size // self.output_stride,
										    )		,    )
										self.parent.assertEqual(
										    result.pooler_output.shape		,    (self.batch_size, self.last_hidden_size)		,    )





						def        SCREAMING_SNAKE_CASE(			self							:Tuple		,    lowerCAmelCase__							:int		,    lowerCAmelCase__							:List[str]		,    lowerCAmelCase__							:str		,    lowerCAmelCase__							:Optional[Any]     )  ->Union[str, Any]:
										lowercase         =      self.num_labels
										lowercase         =      MobileNetVaForImageClassification(lowerCAmelCase__     )
										model.to(lowerCAmelCase__     )
										model.eval()
										lowercase         =      model(lowerCAmelCase__		,    labels=lowerCAmelCase__     )
										self.parent.assertEqual(result.logits.shape		,    (self.batch_size, self.num_labels)     )





						def        SCREAMING_SNAKE_CASE(			self							:Tuple		,    lowerCAmelCase__							:List[str]		,    lowerCAmelCase__							:Any		,    lowerCAmelCase__							:Union[str, Any]		,    lowerCAmelCase__							:Optional[int]     )  ->Union[str, Any]:
										lowercase         =      self.num_labels
										lowercase         =      MobileNetVaForSemanticSegmentation(lowerCAmelCase__     )
										model.to(lowerCAmelCase__     )
										model.eval()
										lowercase         =      model(lowerCAmelCase__     )
										self.parent.assertEqual(
										    result.logits.shape		,    (
										        self.batch_size,
										        self.num_labels,
										        self.image_size // self.output_stride,
										        self.image_size // self.output_stride,
										    )		,    )
										lowercase         =      model(lowerCAmelCase__		,    labels=lowerCAmelCase__     )
										self.parent.assertEqual(
										    result.logits.shape		,    (
										        self.batch_size,
										        self.num_labels,
										        self.image_size // self.output_stride,
										        self.image_size // self.output_stride,
										    )		,    )





						def        SCREAMING_SNAKE_CASE(			self							:Any     )  ->int:
										lowercase         =      self.prepare_config_and_inputs()
										lowercase     ,	lowercase     ,	lowercase     ,	lowercase         =      config_and_inputs
										lowercase         =      {"pixel_values": pixel_values}
										return config, inputs_dict





@require_torch
class   UpperCamelCase_      (      __a		,    __a		,    unittest.TestCase		):







						'''simple docstring'''



						UpperCamelCase      :  Optional[int]				   =					(
						    (MobileNetVaModel, MobileNetVaForImageClassification, MobileNetVaForSemanticSegmentation)
						    if is_torch_available()
						    else ()
						)
						UpperCamelCase      :  List[Any]				   =					(
						    {
						        '''feature-extraction''': MobileNetVaModel,
						        '''image-classification''': MobileNetVaForImageClassification,
						        '''image-segmentation''': MobileNetVaForSemanticSegmentation,
						    }
						    if is_torch_available()
						    else {}
						)

						UpperCamelCase      :  Union[str, Any]				   =					False
						UpperCamelCase      :  int				   =					False
						UpperCamelCase      :  str				   =					False
						UpperCamelCase      :  Dict				   =					False





						def        SCREAMING_SNAKE_CASE(			self							:Optional[int]     )  ->Dict:
										lowercase         =      MobileNetVaModelTester(self     )
										lowercase         =      MobileNetVaConfigTester(self		,    config_class=lowerCAmelCase__		,    has_text_modality=lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:Dict     )  ->Any:
										self.config_tester.run_common_tests()





						@unittest.skip(reason="MobileNetV2 does not use inputs_embeds"     )
						def        SCREAMING_SNAKE_CASE(			self							:Tuple     )  ->Optional[int]:
										pass





						@unittest.skip(reason="MobileNetV2 does not support input and output embeddings"     )
						def        SCREAMING_SNAKE_CASE(			self							:Any     )  ->str:
										pass





						@unittest.skip(reason="MobileNetV2 does not output attentions"     )
						def        SCREAMING_SNAKE_CASE(			self							:List[Any]     )  ->str:
										pass





						def        SCREAMING_SNAKE_CASE(			self							:Union[str, Any]     )  ->Dict:
										lowercase     ,	lowercase         =      self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
														lowercase         =      model_class(lowerCAmelCase__     )
														lowercase         =      inspect.signature(model.forward     )
														# signature.parameters is an OrderedDict => so arg_names order is deterministic
														lowercase         =      [*signature.parameters.keys()]

														lowercase         =      ["pixel_values"]
														self.assertListEqual(arg_names[:1]		,    lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:int     )  ->List[str]:
										lowercase         =      self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_model(*lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:int     )  ->Optional[int]:
										def check_hidden_states_output(lowerCAmelCase__							:str		,    lowerCAmelCase__							:Union[str, Any]		,    lowerCAmelCase__							:int     ):
														lowercase         =      model_class(lowerCAmelCase__     )
														model.to(lowerCAmelCase__     )
														model.eval()

														with torch.no_grad():
																		lowercase         =      model(**self._prepare_for_class(lowerCAmelCase__		,    lowerCAmelCase__     )     )

														lowercase         =      outputs.hidden_states

														lowercase         =      16
														self.assertEqual(len(lowerCAmelCase__     )		,    lowerCAmelCase__     )

										lowercase     ,	lowercase         =      self.model_tester.prepare_config_and_inputs_for_common()

										for model_class in self.all_model_classes:
														lowercase         =      True
														check_hidden_states_output(lowerCAmelCase__		,    lowerCAmelCase__		,    lowerCAmelCase__     )

														# check that output_hidden_states also work using config
														del inputs_dict["output_hidden_states"]
														lowercase         =      True

														check_hidden_states_output(lowerCAmelCase__		,    lowerCAmelCase__		,    lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:Any     )  ->Tuple:
										lowercase         =      self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_image_classification(*lowerCAmelCase__     )





						def        SCREAMING_SNAKE_CASE(			self							:int     )  ->str:
										lowercase         =      self.model_tester.prepare_config_and_inputs()
										self.model_tester.create_and_check_for_semantic_segmentation(*lowerCAmelCase__     )





						@slow
						def        SCREAMING_SNAKE_CASE(			self							:Union[str, Any]     )  ->str:
										for model_name in MOBILENET_V2_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
														lowercase         =      MobileNetVaModel.from_pretrained(lowerCAmelCase__     )
														self.assertIsNotNone(lowerCAmelCase__     )




def  __snake_case							(	):







				'''simple docstring'''


				lowercase         =      Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"						)
				return image





@require_torch
@require_vision
class   UpperCamelCase_      (      unittest.TestCase		):







						'''simple docstring'''



						@cached_property
						def        SCREAMING_SNAKE_CASE(			self							:str     )  ->Any:
										return (
										    MobileNetVaImageProcessor.from_pretrained("google/mobilenet_v2_1.0_224"     ) if is_vision_available() else None
										)





						@slow
						def        SCREAMING_SNAKE_CASE(			self							:Tuple     )  ->Union[str, Any]:
										lowercase         =      MobileNetVaForImageClassification.from_pretrained("google/mobilenet_v2_1.0_224"     ).to(lowerCAmelCase__     )

										lowercase         =      self.default_image_processor
										lowercase         =      prepare_img()
										lowercase         =      image_processor(images=lowerCAmelCase__		,    return_tensors="pt"     ).to(lowerCAmelCase__     )

										# forward pass
										with torch.no_grad():
														lowercase         =      model(**lowerCAmelCase__     )

										# verify the logits
										lowercase         =      torch.Size((1, 1001)     )
										self.assertEqual(outputs.logits.shape		,    lowerCAmelCase__     )

										lowercase         =      torch.tensor([0.24_45, -1.19_93, 0.19_05]     ).to(lowerCAmelCase__     )

										self.assertTrue(torch.allclose(outputs.logits[0, :3]		,    lowerCAmelCase__		,    atol=1E-4     )     )





						@slow
						def        SCREAMING_SNAKE_CASE(			self							:Tuple     )  ->List[Any]:
										lowercase         =      MobileNetVaForSemanticSegmentation.from_pretrained("google/deeplabv3_mobilenet_v2_1.0_513"     )
										lowercase         =      model.to(lowerCAmelCase__     )

										lowercase         =      MobileNetVaImageProcessor.from_pretrained("google/deeplabv3_mobilenet_v2_1.0_513"     )

										lowercase         =      prepare_img()
										lowercase         =      image_processor(images=lowerCAmelCase__		,    return_tensors="pt"     ).to(lowerCAmelCase__     )

										# forward pass
										with torch.no_grad():
														lowercase         =      model(**lowerCAmelCase__     )
										lowercase         =      outputs.logits

										# verify the logits
										lowercase         =      torch.Size((1, 21, 65, 65)     )
										self.assertEqual(logits.shape		,    lowerCAmelCase__     )

										lowercase         =      torch.tensor(
										    [
										        [[17.57_90, 17.75_81, 18.33_55], [18.32_57, 18.42_30, 18.89_73], [18.61_69, 18.86_50, 19.21_87]],
										        [[-2.15_95, -2.09_77, -2.37_41], [-2.42_26, -2.30_28, -2.68_35], [-2.78_19, -2.59_91, -2.77_06]],
										        [[4.20_58, 4.83_17, 4.76_38], [4.41_36, 5.03_61, 4.93_83], [4.50_28, 4.96_44, 4.87_34]],
										    ]		,    device=lowerCAmelCase__		,    )

										self.assertTrue(torch.allclose(logits[0, :3, :3, :3]		,    lowerCAmelCase__		,    atol=1E-4     )     )
