
import inspect
import unittest

from transformers import ViTConfig
from transformers.testing_utils import (
    require_accelerate,
    require_torch,
    require_torch_gpu,
    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 torch import nn

					from transformers import ViTForImageClassification, ViTForMaskedImageModeling, ViTModel
					from transformers.models.vit.modeling_vit import VIT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
					from PIL import Image

					from transformers import ViTImageProcessor







class      A__							:






				def __init__(						self					:      Tuple    ,     a					:      str    ,     a					:      str=13    ,     a					:      Optional[int]=30    ,     a					:      Union[str, Any]=2    ,     a					:      int=3    ,     a					:      Union[str, Any]=True    ,     a					:      Union[str, Any]=True    ,     a					:      List[str]=32    ,     a					:      int=5    ,     a					:      Dict=4    ,     a					:      List[str]=37    ,     a					:      str="gelu"    ,     a					:      int=0.1    ,     a					:      Union[str, Any]=0.1    ,     a					:      Optional[int]=10    ,     a					:      Dict=0.0_2    ,     a					:      Any=None    ,     a					:      Union[str, Any]=2    ,     ):






							'''simple docstring'''







							lowerCAmelCase__      :      Any         =		parent
							lowerCAmelCase__      :      List[Any]         =		batch_size
							lowerCAmelCase__      :      Any         =		image_size
							lowerCAmelCase__      :      Optional[int]         =		patch_size
							lowerCAmelCase__      :      Optional[int]         =		num_channels
							lowerCAmelCase__      :      str         =		is_training
							lowerCAmelCase__      :      Optional[Any]         =		use_labels
							lowerCAmelCase__      :      str         =		hidden_size
							lowerCAmelCase__      :      List[str]         =		num_hidden_layers
							lowerCAmelCase__      :      List[str]         =		num_attention_heads
							lowerCAmelCase__      :      Optional[Any]         =		intermediate_size
							lowerCAmelCase__      :      Dict         =		hidden_act
							lowerCAmelCase__      :      str         =		hidden_dropout_prob
							lowerCAmelCase__      :      Optional[Any]         =		attention_probs_dropout_prob
							lowerCAmelCase__      :      List[str]         =		type_sequence_label_size
							lowerCAmelCase__      :      Any         =		initializer_range
							lowerCAmelCase__      :      int         =		scope
							lowerCAmelCase__      :      Any         =		encoder_stride

							# in ViT, the seq length equals the number of patches + 1 (we add 1 for the [CLS] token)
							lowerCAmelCase__      :      str         =		(image_size // patch_size) ** 2
							lowerCAmelCase__      :      Any         =		num_patches + 1






				def 							_lowerCamelCase		(						self					:      Union[str, Any]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Union[str, Any]         =		floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]   )

							lowerCAmelCase__      :      Tuple         =		None
							if self.use_labels:
										lowerCAmelCase__      :      Optional[Any]         =		ids_tensor([self.batch_size]    ,     self.type_sequence_label_size   )

							lowerCAmelCase__      :      List[str]         =		self.get_config()

							return config, pixel_values, labels






				def 							_lowerCamelCase		(						self					:      Dict   ):






							'''simple docstring'''







							return ViTConfig(
							    image_size=self.image_size    ,     patch_size=self.patch_size    ,     num_channels=self.num_channels    ,     hidden_size=self.hidden_size    ,     num_hidden_layers=self.num_hidden_layers    ,     num_attention_heads=self.num_attention_heads    ,     intermediate_size=self.intermediate_size    ,     hidden_act=self.hidden_act    ,     hidden_dropout_prob=self.hidden_dropout_prob    ,     attention_probs_dropout_prob=self.attention_probs_dropout_prob    ,     is_decoder=a    ,     initializer_range=self.initializer_range    ,     encoder_stride=self.encoder_stride    ,     )






				def 							_lowerCamelCase		(						self					:      Optional[int]    ,     a					:      List[str]    ,     a					:      List[str]    ,     a					:      Union[str, Any]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Tuple         =		ViTModel(config=a   )
							model.to(a   )
							model.eval()
							lowerCAmelCase__      :      int         =		model(a   )
							self.parent.assertEqual(result.last_hidden_state.shape    ,     (self.batch_size, self.seq_length, self.hidden_size)   )






				def 							_lowerCamelCase		(						self					:      Tuple    ,     a					:      Dict    ,     a					:      List[str]    ,     a					:      Tuple   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Optional[int]         =		ViTForMaskedImageModeling(config=a   )
							model.to(a   )
							model.eval()
							lowerCAmelCase__      :      Any         =		model(a   )
							self.parent.assertEqual(
							    result.reconstruction.shape    ,     (self.batch_size, self.num_channels, self.image_size, self.image_size)   )

							# test greyscale images
							lowerCAmelCase__      :      Optional[Any]         =		1
							lowerCAmelCase__      :      List[str]         =		ViTForMaskedImageModeling(a   )
							model.to(a   )
							model.eval()

							lowerCAmelCase__      :      Dict         =		floats_tensor([self.batch_size, 1, self.image_size, self.image_size]   )
							lowerCAmelCase__      :      Union[str, Any]         =		model(a   )
							self.parent.assertEqual(result.reconstruction.shape    ,     (self.batch_size, 1, self.image_size, self.image_size)   )






				def 							_lowerCamelCase		(						self					:      Any    ,     a					:      int    ,     a					:      List[str]    ,     a					:      Union[str, Any]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Any         =		self.type_sequence_label_size
							lowerCAmelCase__      :      Dict         =		ViTForImageClassification(a   )
							model.to(a   )
							model.eval()
							lowerCAmelCase__      :      List[str]         =		model(a    ,     labels=a   )
							self.parent.assertEqual(result.logits.shape    ,     (self.batch_size, self.type_sequence_label_size)   )

							# test greyscale images
							lowerCAmelCase__      :      List[Any]         =		1
							lowerCAmelCase__      :      Tuple         =		ViTForImageClassification(a   )
							model.to(a   )
							model.eval()

							lowerCAmelCase__      :      Dict         =		floats_tensor([self.batch_size, 1, self.image_size, self.image_size]   )
							lowerCAmelCase__      :      Any         =		model(a   )
							self.parent.assertEqual(result.logits.shape    ,     (self.batch_size, self.type_sequence_label_size)   )






				def 							_lowerCamelCase		(						self					:      str   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Tuple         =		self.prepare_config_and_inputs()
							(
							    (
							    lowerCAmelCase__
							)      ,				(
							    lowerCAmelCase__
							)      ,				(
							    lowerCAmelCase__
							)      ,				
							)      :      Any         =		config_and_inputs
							lowerCAmelCase__      :      Optional[int]         =		{'pixel_values': pixel_values}
							return config, inputs_dict







@require_torch
class      A__							(     __magic_name__						,   __magic_name__						,   unittest.TestCase	):
				lowercase			    =  (
				    (
				        ViTModel,
				        ViTForImageClassification,
				        ViTForMaskedImageModeling,
				    )
				    if is_torch_available()
				    else ()
				)
				lowercase			    =  (
				    {'feature-extraction': ViTModel, 'image-classification': ViTForImageClassification}
				    if is_torch_available()
				    else {}
				)
				lowercase			    =  True

				lowercase			    =  False
				lowercase			    =  False
				lowercase			    =  False






				def 							_lowerCamelCase		(						self					:      List[Any]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      List[Any]         =		ViTModelTester(self   )
							lowerCAmelCase__      :      str         =		ConfigTester(self    ,     config_class=a    ,     has_text_modality=a    ,     hidden_size=37   )






				def 							_lowerCamelCase		(						self					:      Any   ):






							'''simple docstring'''







							self.config_tester.run_common_tests()






				@unittest.skip(reason='ViT does not use inputs_embeds'   )
				def 							_lowerCamelCase		(						self					:      str   ):






							'''simple docstring'''







							pass






				def 							_lowerCamelCase		(						self					:      str   ):






							'''simple docstring'''







							lowerCAmelCase__      ,				lowerCAmelCase__      :      str         =		self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
										lowerCAmelCase__      :      Union[str, Any]         =		model_class(a   )
										self.assertIsInstance(model.get_input_embeddings()    ,     (nn.Module)   )
										lowerCAmelCase__      :      Optional[Any]         =		model.get_output_embeddings()
										self.assertTrue(x is None or isinstance(a    ,     nn.Linear   )   )






				def 							_lowerCamelCase		(						self					:      int   ):






							'''simple docstring'''







							lowerCAmelCase__      ,				lowerCAmelCase__      :      List[str]         =		self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
										lowerCAmelCase__      :      Union[str, Any]         =		model_class(a   )
										lowerCAmelCase__      :      Dict         =		inspect.signature(model.forward   )
										# signature.parameters is an OrderedDict => so arg_names order is deterministic
										lowerCAmelCase__      :      int         =		[*signature.parameters.keys()]

										lowerCAmelCase__      :      str         =		['pixel_values']
										self.assertListEqual(arg_names[:1]    ,     a   )






				def 							_lowerCamelCase		(						self					:      Dict   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Any         =		self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*a   )






				def 							_lowerCamelCase		(						self					:      List[Any]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Optional[Any]         =		self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_masked_image_modeling(*a   )






				def 							_lowerCamelCase		(						self					:      Dict   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Dict         =		self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_image_classification(*a   )






				@slow
				def 							_lowerCamelCase		(						self					:      str   ):






							'''simple docstring'''







							for model_name in VIT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
										lowerCAmelCase__      :      Optional[Any]         =		ViTModel.from_pretrained(a   )
										self.assertIsNotNone(a   )






def  lowerCAmelCase__			(  )				->		Union[str, Any]:
			lowerCAmelCase__      :      Dict         =		Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'     )
			return image







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






				@cached_property
				def 							_lowerCamelCase		(						self					:      List[Any]   ):






							'''simple docstring'''







							return ViTImageProcessor.from_pretrained('google/vit-base-patch16-224'   ) if is_vision_available() else None






				@slow
				def 							_lowerCamelCase		(						self					:      Optional[int]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Tuple         =		ViTForImageClassification.from_pretrained('google/vit-base-patch16-224'   ).to(a   )

							lowerCAmelCase__      :      Optional[int]         =		self.default_image_processor
							lowerCAmelCase__      :      Union[str, Any]         =		prepare_img()
							lowerCAmelCase__      :      Optional[Any]         =		image_processor(images=a    ,     return_tensors='pt'   ).to(a   )

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

							# verify the logits
							lowerCAmelCase__      :      List[str]         =		torch.Size((1, 1_000)   )
							self.assertEqual(outputs.logits.shape    ,     a   )

							lowerCAmelCase__      :      Any         =		torch.tensor([-0.2_7_4_4, 0.8_2_1_5, -0.0_8_3_6]   ).to(a   )

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






				@slow
				def 							_lowerCamelCase		(						self					:      List[str]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      Any         =		ViTModel.from_pretrained('facebook/dino-vits8'   ).to(a   )

							lowerCAmelCase__      :      Dict         =		ViTImageProcessor.from_pretrained('facebook/dino-vits8'    ,     size=480   )
							lowerCAmelCase__      :      Any         =		prepare_img()
							lowerCAmelCase__      :      str         =		image_processor(images=a    ,     return_tensors='pt'   )
							lowerCAmelCase__      :      str         =		inputs.pixel_values.to(a   )

							# forward pass
							with torch.no_grad():
										lowerCAmelCase__      :      List[str]         =		model(a    ,     interpolate_pos_encoding=a   )

							# verify the logits
							lowerCAmelCase__      :      Union[str, Any]         =		torch.Size((1, 3_601, 384)   )
							self.assertEqual(outputs.last_hidden_state.shape    ,     a   )

							lowerCAmelCase__      :      Tuple         =		torch.tensor(
							    [[4.2_3_4_0, 4.3_9_0_6, -6.6_6_9_2], [4.5_4_6_3, 1.8_9_2_8, -6.7_2_5_7], [4.4_4_2_9, 0.8_4_9_6, -5.8_5_8_5]]   ).to(a   )

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






				@slow
				@require_accelerate
				@require_torch_gpu
				def 							_lowerCamelCase		(						self					:      Optional[Any]   ):






							'''simple docstring'''







							lowerCAmelCase__      :      int         =		ViTModel.from_pretrained('facebook/dino-vits8'    ,     torch_dtype=torch.floataa    ,     device_map='auto'   )
							lowerCAmelCase__      :      List[str]         =		self.default_image_processor

							lowerCAmelCase__      :      Tuple         =		prepare_img()
							lowerCAmelCase__      :      List[str]         =		image_processor(images=a    ,     return_tensors='pt'   )
							lowerCAmelCase__      :      Dict         =		inputs.pixel_values.to(a   )

							# forward pass to make sure inference works in fp16
							with torch.no_grad():
										lowerCAmelCase__      :      Dict         =		model(a   )

import datetime
import platform
import subprocess
from typing import Optional, Tuple, Union

import numpy as np
def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )				->		np.array:
			lowerCAmelCase__      :      Dict         =		F'''{sampling_rate}'''
			lowerCAmelCase__      :      Any         =		'1'
			lowerCAmelCase__      :      Optional[Any]         =		'f32le'
			lowerCAmelCase__      :      Any         =		[
			    'ffmpeg',
			    '-i',
			    'pipe:0',
			    '-ac',
			    ac,
			    '-ar',
			    ar,
			    '-f',
			    format_for_conversion,
			    '-hide_banner',
			    '-loglevel',
			    'quiet',
			    'pipe:1',
			]

			try:
						with subprocess.Popen(SCREAMING_SNAKE_CASE_       ,							stdin=subprocess.PIPE       ,							stdout=subprocess.PIPE     ) as ffmpeg_process:
									lowerCAmelCase__      :      List[Any]         =		ffmpeg_process.communicate(SCREAMING_SNAKE_CASE_     )
			except FileNotFoundError as error:
						raise ValueError('ffmpeg was not found but is required to load audio files from filename'     ) from error
			lowerCAmelCase__      :      List[str]         =		output_stream[0]
			lowerCAmelCase__      :      str         =		np.frombuffer(SCREAMING_SNAKE_CASE_       ,							np.floataa     )
			if audio.shape[0] == 0:
						raise ValueError('Malformed soundfile'     )
			return audio
def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_ = "f32le"       ,							)				->		Dict:
			lowerCAmelCase__      :      Optional[Any]         =		F'''{sampling_rate}'''
			lowerCAmelCase__      :      Any         =		'1'
			if format_for_conversion == "s16le":
						lowerCAmelCase__      :      Dict         =		2
			elif format_for_conversion == "f32le":
						lowerCAmelCase__      :      List[str]         =		4
			else:
						raise ValueError(F'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''     )

			lowerCAmelCase__      :      Tuple         =		platform.system()
			if system == "Linux":
						lowerCAmelCase__      :      str         =		'alsa'
						lowerCAmelCase__      :      str         =		'default'
			elif system == "Darwin":
						lowerCAmelCase__      :      Any         =		'avfoundation'
						lowerCAmelCase__      :      Tuple         =		':0'
			elif system == "Windows":
						lowerCAmelCase__      :      Any         =		'dshow'
						lowerCAmelCase__      :      int         =		'default'

			lowerCAmelCase__      :      Any         =		[
			    'ffmpeg',
			    '-f',
			    format_,
			    '-i',
			    input_,
			    '-ac',
			    ac,
			    '-ar',
			    ar,
			    '-f',
			    format_for_conversion,
			    '-fflags',
			    'nobuffer',
			    '-hide_banner',
			    '-loglevel',
			    'quiet',
			    'pipe:1',
			]
			lowerCAmelCase__      :      List[str]         =		int(round(sampling_rate * chunk_length_s     )     ) * size_of_sample
			lowerCAmelCase__      :      str         =		_ffmpeg_stream(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )
			for item in iterator:
						yield item
def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_ = None       ,							SCREAMING_SNAKE_CASE_ = None       ,							SCREAMING_SNAKE_CASE_ = "f32le"       ,							)				->		str:
			if stream_chunk_s is not None:
						lowerCAmelCase__      :      Union[str, Any]         =		stream_chunk_s
			else:
						lowerCAmelCase__      :      Tuple         =		chunk_length_s

			lowerCAmelCase__      :      Any         =		ffmpeg_microphone(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							format_for_conversion=SCREAMING_SNAKE_CASE_     )
			if format_for_conversion == "s16le":
						lowerCAmelCase__      :      Optional[Any]         =		np.intaa
						lowerCAmelCase__      :      Optional[Any]         =		2
			elif format_for_conversion == "f32le":
						lowerCAmelCase__      :      Optional[Any]         =		np.floataa
						lowerCAmelCase__      :      Optional[Any]         =		4
			else:
						raise ValueError(F'''Unhandled format `{format_for_conversion}`. Please use `s16le` or `f32le`'''     )

			if stride_length_s is None:
						lowerCAmelCase__      :      Dict         =		chunk_length_s / 6
			lowerCAmelCase__      :      int         =		int(round(sampling_rate * chunk_length_s     )     ) * size_of_sample
			if isinstance(SCREAMING_SNAKE_CASE_       ,							(int, float)     ):
						lowerCAmelCase__      :      Dict         =		[stride_length_s, stride_length_s]

			lowerCAmelCase__      :      Union[str, Any]         =		int(round(sampling_rate * stride_length_s[0]     )     ) * size_of_sample
			lowerCAmelCase__      :      List[Any]         =		int(round(sampling_rate * stride_length_s[1]     )     ) * size_of_sample
			lowerCAmelCase__      :      Any         =		datetime.datetime.now()
			lowerCAmelCase__      :      Any         =		datetime.timedelta(seconds=SCREAMING_SNAKE_CASE_     )
			for item in chunk_bytes_iter(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							stride=(stride_left, stride_right)       ,							stream=SCREAMING_SNAKE_CASE_     ):
						# Put everything back in numpy scale
						lowerCAmelCase__      :      Any         =		np.frombuffer(item['raw']       ,							dtype=SCREAMING_SNAKE_CASE_     )
						lowerCAmelCase__      :      Optional[Any]         =		(
						    item['stride'][0] // size_of_sample,
						    item['stride'][1] // size_of_sample,
						)
						lowerCAmelCase__      :      Optional[int]         =		sampling_rate
						audio_time += delta
						if datetime.datetime.now() > audio_time + 10 * delta:
									# We're late !! SKIP
									continue
						yield item
def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_ = False     )				->		Optional[int]:
			lowerCAmelCase__      :      Union[str, Any]         =		b''
			lowerCAmelCase__      ,				lowerCAmelCase__      :      Union[str, Any]         =		stride
			if stride_left + stride_right >= chunk_len:
						raise ValueError(
						    F'''Stride needs to be strictly smaller than chunk_len: ({stride_left}, {stride_right}) vs {chunk_len}'''     )
			lowerCAmelCase__      :      List[str]         =		0
			for raw in iterator:
						acc += raw
						if stream and len(SCREAMING_SNAKE_CASE_     ) < chunk_len:
									lowerCAmelCase__      :      Tuple         =		(_stride_left, 0)
									yield {"raw": acc[:chunk_len], "stride": stride, "partial": True}
						else:
									while len(SCREAMING_SNAKE_CASE_     ) >= chunk_len:
												# We are flushing the accumulator
												lowerCAmelCase__      :      Dict         =		(_stride_left, stride_right)
												lowerCAmelCase__      :      Any         =		{'raw': acc[:chunk_len], 'stride': stride}
												if stream:
															lowerCAmelCase__      :      Optional[int]         =		False
												yield item
												lowerCAmelCase__      :      Optional[int]         =		stride_left
												lowerCAmelCase__      :      Optional[int]         =		acc[chunk_len - stride_left - stride_right :]
    # Last chunk
			if len(SCREAMING_SNAKE_CASE_     ) > stride_left:
						lowerCAmelCase__      :      Tuple         =		{'raw': acc, 'stride': (_stride_left, 0)}
						if stream:
									lowerCAmelCase__      :      Any         =		False
						yield item







def  lowerCAmelCase__			(  SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_     )				->		str:
			lowerCAmelCase__      :      int         =		2**24  # 16Mo
			try:
						with subprocess.Popen(SCREAMING_SNAKE_CASE_       ,							stdout=subprocess.PIPE       ,							bufsize=SCREAMING_SNAKE_CASE_     ) as ffmpeg_process:
									while True:
												lowerCAmelCase__      :      List[str]         =		ffmpeg_process.stdout.read(SCREAMING_SNAKE_CASE_     )
												if raw == b"":
															break
												yield raw
			except FileNotFoundError as error:
						raise ValueError('ffmpeg was not found but is required to stream audio files from filename'     ) from error