


import copy
import inspect
import unittest

from transformers import PretrainedConfig, SwiftFormerConfig
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 torch import nn

			from transformers import SwiftFormerForImageClassification, SwiftFormerModel
			from transformers.models.swiftformer.modeling_swiftformer import SWIFTFORMER_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
			from PIL import Image

			from transformers import ViTImageProcessor

class 				a		:




							"""simple docstring"""


							def __init__(							self       :  List[Any]    ,	lowerCamelCase       :  Any    ,	lowerCamelCase       :  Dict=13    ,	lowerCamelCase       :  Optional[Any]=3    ,	lowerCamelCase       :  Optional[Any]=True    ,	lowerCamelCase       :  Union[str, Any]=True    ,	lowerCamelCase       :  Union[str, Any]=0.1    ,	lowerCamelCase       :  List[Any]=0.1    ,	lowerCamelCase       :  List[str]=224    ,	lowerCamelCase       :  List[str]=1000    ,	lowerCamelCase       :  List[str]=[3, 3, 6, 4]    ,	lowerCamelCase       :  Optional[Any]=[48, 56, 112, 220]    ,	)	->    Union[str, Any]:
														__snake_case		:			Dict							 =   parent
														__snake_case		:			List[Any]							 =   batch_size
														__snake_case		:			Dict							 =   num_channels
														__snake_case		:			Optional[Any]							 =   is_training
														__snake_case		:			List[str]							 =   use_labels
														__snake_case		:			List[Any]							 =   hidden_dropout_prob
														__snake_case		:			Optional[int]							 =   attention_probs_dropout_prob
														__snake_case		:			Tuple							 =   num_labels
														__snake_case		:			List[str]							 =   image_size
														__snake_case		:			str							 =   layer_depths
														__snake_case		:			Optional[int]							 =   embed_dims

							def     __snake_case					(							self       :  int				)	->    Optional[int]:
														__snake_case		:			Union[str, Any]							 =   floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]				)

														__snake_case		:			int							 =   None
														if self.use_labels:
																					__snake_case		:			Tuple							 =   ids_tensor([self.batch_size]    ,	self.num_labels				)

														__snake_case		:			int							 =   self.get_config()

														return config, pixel_values, labels

							def     __snake_case					(							self       :  str				)	->    List[str]:
														return SwiftFormerConfig(
														    depths=self.layer_depths    ,	embed_dims=self.embed_dims    ,	mlp_ratio=4    ,	downsamples=[True, True, True, True]    ,	hidden_act="gelu"    ,	num_labels=self.num_labels    ,	down_patch_size=3    ,	down_stride=2    ,	down_pad=1    ,	drop_rate=0.0    ,	drop_path_rate=0.0    ,	use_layer_scale=lowerCamelCase    ,	layer_scale_init_value=1E-5    ,	)

							def     __snake_case					(							self       :  List[Any]    ,	lowerCamelCase       :  Tuple    ,	lowerCamelCase       :  str    ,	lowerCamelCase       :  Union[str, Any]				)	->    Tuple:
														__snake_case		:			List[Any]							 =   SwiftFormerModel(config=lowerCamelCase				)
														model.to(lowerCamelCase				)
														model.eval()
														__snake_case		:			Union[str, Any]							 =   model(lowerCamelCase				)
														self.parent.assertEqual(result.last_hidden_state.shape    ,	(self.batch_size, self.embed_dims[-1], 7, 7)				)

							def     __snake_case					(							self       :  Optional[int]    ,	lowerCamelCase       :  Dict    ,	lowerCamelCase       :  List[Any]    ,	lowerCamelCase       :  List[str]				)	->    Dict:
														__snake_case		:			Any							 =   self.num_labels
														__snake_case		:			Any							 =   SwiftFormerForImageClassification(lowerCamelCase				)
														model.to(lowerCamelCase				)
														model.eval()
														__snake_case		:			Optional[int]							 =   model(lowerCamelCase    ,	labels=lowerCamelCase				)
														self.parent.assertEqual(result.logits.shape    ,	(self.batch_size, self.num_labels)				)

														__snake_case		:			int							 =   SwiftFormerForImageClassification(lowerCamelCase				)
														model.to(lowerCamelCase				)
														model.eval()

														__snake_case		:			int							 =   floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]				)
														__snake_case		:			Dict							 =   model(lowerCamelCase				)
														self.parent.assertEqual(result.logits.shape    ,	(self.batch_size, self.num_labels)				)






							def     __snake_case					(							self       :  Optional[Any]				)	->    Tuple:
														(__snake_case)		:			int							 =   self.prepare_config_and_inputs()
														__snake_case		:			Any							 =   {"""pixel_values""": pixel_values}
														return config, inputs_dict

@require_torch
class 				a		(__UpperCAmelCase     ,				__UpperCAmelCase     ,				unittest.TestCase      ):




							"""simple docstring"""


							__UpperCAmelCase    :							Optional[Any]       =  (SwiftFormerModel, SwiftFormerForImageClassification) if is_torch_available() else ()
							__UpperCAmelCase    :							Optional[Any]       =  (
							    {'feature-extraction': SwiftFormerModel, 'image-classification': SwiftFormerForImageClassification}
							    if is_torch_available()
							    else {}
							)

							__UpperCAmelCase    :							int       =  False
							__UpperCAmelCase    :							List[Any]       =  False
							__UpperCAmelCase    :							List[str]       =  False
							__UpperCAmelCase    :							Union[str, Any]       =  False
							__UpperCAmelCase    :							Union[str, Any]       =  False

							def     __snake_case					(							self       :  Dict				)	->    Optional[Any]:
														__snake_case		:			Optional[int]							 =   SwiftFormerModelTester(self				)
														__snake_case		:			Any							 =   ConfigTester(
														    self    ,	config_class=lowerCamelCase    ,	has_text_modality=lowerCamelCase    ,	hidden_size=37    ,	num_attention_heads=12    ,	num_hidden_layers=12    ,	)

							def     __snake_case					(							self       :  str				)	->    Any:
														self.config_tester.run_common_tests()

							@unittest.skip(reason="SwiftFormer does not use inputs_embeds"				)
							def     __snake_case					(							self       :  int				)	->    int:
														pass

							def     __snake_case					(							self       :  str				)	->    Any:
														__snake_case		:			Tuple							 =   self.model_tester.prepare_config_and_inputs_for_common()

														for model_class in self.all_model_classes:
																					__snake_case		:			Optional[Any]							 =   model_class(lowerCamelCase				)
																					__snake_case		:			Optional[int]							 =   model.get_output_embeddings()
																					self.assertTrue(x is None or isinstance(lowerCamelCase    ,	nn.Linear				)				)

							def     __snake_case					(							self       :  Dict				)	->    Union[str, Any]:
														__snake_case		:			Optional[int]							 =   self.model_tester.prepare_config_and_inputs_for_common()

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

																					__snake_case		:			Tuple							 =   ["""pixel_values"""]
																					self.assertListEqual(arg_names[:1]    ,	lowerCamelCase				)

							def     __snake_case					(							self       :  str				)	->    int:
														__snake_case		:			str							 =   self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_model(*lowerCamelCase				)

							def     __snake_case					(							self       :  Optional[int]				)	->    Dict:
														__snake_case		:			List[Any]							 =   self.model_tester.prepare_config_and_inputs()
														self.model_tester.create_and_check_for_image_classification(*lowerCamelCase				)

							@slow
							def     __snake_case					(							self       :  List[Any]				)	->    Optional[Any]:
														for model_name in SWIFTFORMER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																					__snake_case		:			Optional[Any]							 =   SwiftFormerModel.from_pretrained(lowerCamelCase				)
																					self.assertIsNotNone(lowerCamelCase				)

							@unittest.skip(reason="SwiftFormer does not output attentions"				)
							def     __snake_case					(							self       :  List[Any]				)	->    Tuple:
														pass

							def     __snake_case					(							self       :  str				)	->    int:
														def check_hidden_states_output(lowerCamelCase       :  Optional[Any]    ,	lowerCamelCase       :  List[str]    ,	lowerCamelCase       :  Union[str, Any]				):
																					__snake_case		:			str							 =   model_class(lowerCamelCase				)
																					model.to(lowerCamelCase				)
																					model.eval()

																					with torch.no_grad():
																												__snake_case		:			Optional[int]							 =   model(**self._prepare_for_class(lowerCamelCase    ,	lowerCamelCase				)				)

																					__snake_case		:			Any							 =   outputs.hidden_states

																					__snake_case		:			Any							 =   8
																					self.assertEqual(len(lowerCamelCase				)    ,	lowerCamelCase				)  # TODO

																					# SwiftFormer's feature maps are of shape (batch_size, embed_dims, height, width)
																					# with the width and height being successively divided by 2, after every 2 blocks
																					for i in range(len(lowerCamelCase				)				):
																												self.assertEqual(
																												    hidden_states[i].shape    ,	torch.Size(
																												        [
																												            self.model_tester.batch_size,
																												            self.model_tester.embed_dims[i // 2],
																												            (self.model_tester.image_size // 4) // 2 ** (i // 2),
																												            (self.model_tester.image_size // 4) // 2 ** (i // 2),
																												        ]				)    ,	)

														__snake_case		:			str							 =   self.model_tester.prepare_config_and_inputs_for_common()

														for model_class in self.all_model_classes:
																					__snake_case		:			str							 =   True
																					check_hidden_states_output(lowerCamelCase    ,	lowerCamelCase    ,	lowerCamelCase				)

																					# check that output_hidden_states also work using config
																					del inputs_dict["output_hidden_states"]
																					__snake_case		:			str							 =   True

																					check_hidden_states_output(lowerCamelCase    ,	lowerCamelCase    ,	lowerCamelCase				)

							def     __snake_case					(							self       :  Optional[int]				)	->    Tuple:
														def _config_zero_init(lowerCamelCase       :  int				):
																					__snake_case		:			Optional[Any]							 =   copy.deepcopy(lowerCamelCase				)
																					for key in configs_no_init.__dict__.keys():
																												if "_range" in key or "_std" in key or "initializer_factor" in key or "layer_scale" in key:
																																			setattr(lowerCamelCase    ,	lowerCamelCase    ,	1E-10				)
																												if isinstance(getattr(lowerCamelCase    ,	lowerCamelCase    ,	lowerCamelCase				)    ,	lowerCamelCase				):
																																			__snake_case		:			Any							 =   _config_zero_init(getattr(lowerCamelCase    ,	lowerCamelCase				)				)
																																			setattr(lowerCamelCase    ,	lowerCamelCase    ,	lowerCamelCase				)
																					return configs_no_init

														__snake_case		:			Optional[int]							 =   self.model_tester.prepare_config_and_inputs_for_common()

														__snake_case		:			Any							 =   _config_zero_init(lowerCamelCase				)
														for model_class in self.all_model_classes:
																					__snake_case		:			List[str]							 =   model_class(config=lowerCamelCase				)
																					for name, param in model.named_parameters():
																												if param.requires_grad:
																																			self.assertIn(
																																			    ((param.data.mean() * 1E9) / 1E9).round().item()    ,	[0.0, 1.0]    ,	msg=F'Parameter {name} of model {model_class} seems not properly initialized'    ,	)






							@unittest.skip("Will be fixed soon by reducing the size of the model used for common tests."				)
							def     __snake_case					(							self       :  Optional[int]				)	->    List[Any]:
														pass







def 				lowerCAmelCase_     (					):
							__snake_case		:			List[str]							 =   Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"	)
							return image

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




							"""simple docstring"""


							@cached_property
							def     __snake_case					(							self       :  Tuple				)	->    Optional[Any]:
														return ViTImageProcessor.from_pretrained("MBZUAI/swiftformer-xs"				) if is_vision_available() else None






							@slow
							def     __snake_case					(							self       :  List[str]				)	->    Union[str, Any]:
														__snake_case		:			Any							 =   SwiftFormerForImageClassification.from_pretrained("MBZUAI/swiftformer-xs"				).to(lowerCamelCase				)

														__snake_case		:			Dict							 =   self.default_image_processor
														__snake_case		:			Dict							 =   prepare_img()
														__snake_case		:			int							 =   image_processor(images=lowerCamelCase    ,	return_tensors="pt"				).to(lowerCamelCase				)

														# forward pass
														with torch.no_grad():
																					__snake_case		:			int							 =   model(**lowerCamelCase				)

														# verify the logits
														__snake_case		:			int							 =   torch.Size((1, 1000)				)
														self.assertEqual(outputs.logits.shape    ,	lowerCamelCase				)

														__snake_case		:			List[str]							 =   torch.tensor([[-2.1703E00, 2.1107E00, -2.0811E00]]				).to(lowerCamelCase				)
														self.assertTrue(torch.allclose(outputs.logits[0, :3]    ,	lowerCamelCase    ,	atol=1E-4				)				)






'''simple docstring'''




from collections import UserDict
from typing import Union

import numpy as np
import requests

from ..utils import (
    add_end_docstrings,
    logging,
)
from .audio_classification import ffmpeg_read
from .base import PIPELINE_INIT_ARGS, Pipeline


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







@add_end_docstrings(__UpperCAmelCase )
class 	_lowerCAmelCase							(			__UpperCAmelCase ):


	def __init__(self       ,      **lowercase							):
				super().__init__(**lowercase							)

				if self.framework != "pt":
							raise ValueError(F'The {self.__class__} is only available in PyTorch.'							)
				# No specific FOR_XXX available yet


	def __call__(self       ,      lowercase       ,      **lowercase							):
				return super().__call__(lowercase       ,      **lowercase							)


	def     _a	(self       ,      **lowercase							):
				A_	:						Tuple	=     {}
				if "candidate_labels" in kwargs:
							A_	:						Dict	=     kwargs["""candidate_labels"""]
				if "hypothesis_template" in kwargs:
							A_	:						Optional[Any]	=     kwargs["""hypothesis_template"""]

				return preprocess_params, {}, {}


	def     _a	(self       ,      lowercase       ,      lowercase=None       ,      lowercase="This is a sound of {}."							):
				if isinstance(lowercase       ,      lowercase							):
							if audio.startswith("""http://"""							) or audio.startswith("""https://"""							):
										# We need to actually check for a real protocol, otherwise it's impossible to use a local file
										# like http_huggingface_co.png
										A_	:						Dict	=     requests.get(lowercase							).content
							else:
										with open(lowercase       ,      """rb"""							) as f:
													A_	:						List[str]	=     f.read()

				if isinstance(lowercase       ,      lowercase							):
							A_	:						List[Any]	=     ffmpeg_read(lowercase       ,      self.feature_extractor.sampling_rate							)

				if not isinstance(lowercase       ,      np.ndarray							):
							raise ValueError("""We expect a numpy ndarray as input"""							)
				if len(audio.shape							) != 1:
							raise ValueError("""We expect a single channel audio input for ZeroShotAudioClassificationPipeline"""							)

				A_	:						int	=     self.feature_extractor(
				    [audio]       ,      sampling_rate=self.feature_extractor.sampling_rate       ,      return_tensors="""pt"""							)
				A_	:						List[Any]	=     candidate_labels
				A_	:						str	=     [hypothesis_template.format(lowercase							) for x in candidate_labels]
				A_	:						Optional[Any]	=     self.tokenizer(lowercase       ,      return_tensors=self.framework       ,      padding=lowercase							)
				A_	:						Optional[Any]	=     [text_inputs]
				return inputs


	def     _a	(self       ,      lowercase							):
				A_	:						Union[str, Any]	=     model_inputs.pop("""candidate_labels"""							)
				A_	:						List[Any]	=     model_inputs.pop("""text_inputs"""							)
				if isinstance(text_inputs[0]       ,      lowercase							):
							A_	:						Union[str, Any]	=     text_inputs[0]
				else:
							# Batching case.
							A_	:						Optional[int]	=     text_inputs[0][0]

				A_	:						str	=     self.model(**lowercase       ,      **lowercase							)

				A_	:						Union[str, Any]	=     {
				    """candidate_labels""": candidate_labels,
				    """logits""": outputs.logits_per_audio,
				}
				return model_outputs


	def     _a	(self       ,      lowercase							):
				A_	:						Union[str, Any]	=     model_outputs.pop("""candidate_labels"""							)
				A_	:						List[Any]	=     model_outputs["""logits"""][0]

				if self.framework == "pt":
							A_	:						Optional[Any]	=     logits.softmax(dim=0							)
							A_	:						str	=     probs.tolist()
				else:
							raise ValueError("""`tf` framework not supported."""							)

				A_	:						Optional[int]	=     [
				    {"""score""": score, """label""": candidate_label}
				    for score, candidate_label in sorted(zip(lowercase       ,      lowercase							)       ,      key=lambda lowercase							: -x[0]							)
				]
				return result