
"""simple docstring"""
import collections
import inspect
import unittest

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

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


if is_torch_available():
			import torch
			from torch import nn

			from transformers import DinatBackbone, DinatForImageClassification, DinatModel
			from transformers.models.dinat.modeling_dinat import DINAT_PRETRAINED_MODEL_ARCHIVE_LIST

if is_vision_available():
			from PIL import Image

			from transformers import AutoImageProcessor


class    _lowerCAmelCase						:



							def __init__(      self ,						a_ ,						a_=13 ,						a_=64 ,						a_=4 ,						a_=3 ,						a_=16 ,						a_=[1, 2, 1] ,						a_=[2, 4, 8] ,						a_=3 ,						a_=[[3], [1, 2], [1]] ,						a_=2.0 ,						a_=True ,						a_=0.0 ,						a_=0.0 ,						a_=0.1 ,						a_="gelu" ,						a_=True ,						a_=0.02 ,						a_=1e-5 ,						a_=True ,						a_=None ,						a_=True ,						a_=10 ,						a_=["stage1", "stage2"] ,						a_=[1, 2] ,						)							->							int:
								_UpperCAmelCase =						parent
								_UpperCAmelCase =						batch_size
								_UpperCAmelCase =						image_size
								_UpperCAmelCase =						patch_size
								_UpperCAmelCase =						num_channels
								_UpperCAmelCase =						embed_dim
								_UpperCAmelCase =						depths
								_UpperCAmelCase =						num_heads
								_UpperCAmelCase =						kernel_size
								_UpperCAmelCase =						dilations
								_UpperCAmelCase =						mlp_ratio
								_UpperCAmelCase =						qkv_bias
								_UpperCAmelCase =						hidden_dropout_prob
								_UpperCAmelCase =						attention_probs_dropout_prob
								_UpperCAmelCase =						drop_path_rate
								_UpperCAmelCase =						hidden_act
								_UpperCAmelCase =						patch_norm
								_UpperCAmelCase =						layer_norm_eps
								_UpperCAmelCase =						initializer_range
								_UpperCAmelCase =						is_training
								_UpperCAmelCase =						scope
								_UpperCAmelCase =						use_labels
								_UpperCAmelCase =						num_labels
								_UpperCAmelCase =						out_features
								_UpperCAmelCase =						out_indices



							def       _a   (      self   )							->							Union[str, Any]:
								_UpperCAmelCase =						floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]   )

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

								_UpperCAmelCase =						self.get_config()

								return config, pixel_values, labels



							def       _a   (      self   )							->							List[Any]:
								return DinatConfig(
								    num_labels=self.num_labels ,						image_size=self.image_size ,						patch_size=self.patch_size ,						num_channels=self.num_channels ,						embed_dim=self.embed_dim ,						depths=self.depths ,						num_heads=self.num_heads ,						kernel_size=self.kernel_size ,						dilations=self.dilations ,						mlp_ratio=self.mlp_ratio ,						qkv_bias=self.qkv_bias ,						hidden_dropout_prob=self.hidden_dropout_prob ,						attention_probs_dropout_prob=self.attention_probs_dropout_prob ,						drop_path_rate=self.drop_path_rate ,						hidden_act=self.hidden_act ,						patch_norm=self.patch_norm ,						layer_norm_eps=self.layer_norm_eps ,						initializer_range=self.initializer_range ,						out_features=self.out_features ,						out_indices=self.out_indices ,						)



							def       _a   (      self ,						a_ ,						a_ ,						a_   )							->							int:
								_UpperCAmelCase =						DinatModel(config=a_   )
								model.to(a_   )
								model.eval()
								_UpperCAmelCase =						model(a_   )

								_UpperCAmelCase =						_UpperCAmelCase =						(config.image_size // config.patch_size) // (2 ** (len(config.depths   ) - 1))
								_UpperCAmelCase =						int(config.embed_dim * 2 ** (len(config.depths   ) - 1)   )

								self.parent.assertEqual(
								    result.last_hidden_state.shape ,						(self.batch_size, expected_height, expected_width, expected_dim)   )



							def       _a   (      self ,						a_ ,						a_ ,						a_   )							->							Any:
								_UpperCAmelCase =						DinatForImageClassification(a_   )
								model.to(a_   )
								model.eval()
								_UpperCAmelCase =						model(a_ ,						labels=a_   )
								self.parent.assertEqual(result.logits.shape ,						(self.batch_size, self.num_labels)   )

								# test greyscale images
								_UpperCAmelCase =						1
								_UpperCAmelCase =						DinatForImageClassification(a_   )
								model.to(a_   )
								model.eval()

								_UpperCAmelCase =						floats_tensor([self.batch_size, 1, self.image_size, self.image_size]   )
								_UpperCAmelCase =						model(a_   )
								self.parent.assertEqual(result.logits.shape ,						(self.batch_size, self.num_labels)   )



							def       _a   (      self ,						a_ ,						a_ ,						a_   )							->							Dict:
								_UpperCAmelCase =						DinatBackbone(config=a_   )
								model.to(a_   )
								model.eval()
								_UpperCAmelCase =						model(a_   )

								# verify hidden states
								self.parent.assertEqual(len(result.feature_maps   ) ,						len(config.out_features   )   )
								self.parent.assertListEqual(list(result.feature_maps[0].shape   ) ,						[self.batch_size, model.channels[0], 16, 16]   )

								# verify channels
								self.parent.assertEqual(len(model.channels   ) ,						len(config.out_features   )   )

								# verify backbone works with out_features=None
								_UpperCAmelCase =						None
								_UpperCAmelCase =						DinatBackbone(config=a_   )
								model.to(a_   )
								model.eval()
								_UpperCAmelCase =						model(a_   )

								# verify feature maps
								self.parent.assertEqual(len(result.feature_maps   ) ,						1   )
								self.parent.assertListEqual(list(result.feature_maps[0].shape   ) ,						[self.batch_size, model.channels[-1], 4, 4]   )

								# verify channels
								self.parent.assertEqual(len(model.channels   ) ,						1   )



							def       _a   (      self   )							->							Optional[int]:
								_UpperCAmelCase =						self.prepare_config_and_inputs()
								_UpperCAmelCase ,      _UpperCAmelCase ,      _UpperCAmelCase =						config_and_inputs
								_UpperCAmelCase =						{"pixel_values": pixel_values}
								return config, inputs_dict





@require_natten
@require_torch
class    _lowerCAmelCase						(						lowerCamelCase    ,       lowerCamelCase    ,       unittest.TestCase						):
							lowercase_     :     int       =							(
							    (
							        DinatModel,
							        DinatForImageClassification,
							        DinatBackbone,
							    )
							    if is_torch_available()
							    else ()
							)
							lowercase_     :     List[Any]       =							(
							    {'''feature-extraction''': DinatModel, '''image-classification''': DinatForImageClassification}
							    if is_torch_available()
							    else {}
							)
							lowercase_     :     List[Any]       =							False

							lowercase_     :     Optional[int]       =							False
							lowercase_     :     Dict       =							False
							lowercase_     :     Optional[Any]       =							False
							lowercase_     :     List[str]       =							False



							def       _a   (      self   )							->							Tuple:
								_UpperCAmelCase =						DinatModelTester(self   )
								_UpperCAmelCase =						ConfigTester(self ,						config_class=a_ ,						embed_dim=37   )



							def       _a   (      self   )							->							int:
								self.create_and_test_config_common_properties()
								self.config_tester.create_and_test_config_to_json_string()
								self.config_tester.create_and_test_config_to_json_file()
								self.config_tester.create_and_test_config_from_and_save_pretrained()
								self.config_tester.create_and_test_config_with_num_labels()
								self.config_tester.check_config_can_be_init_without_params()
								self.config_tester.check_config_arguments_init()



							def       _a   (      self   )							->							Dict:
								return



							def       _a   (      self   )							->							str:
								_UpperCAmelCase =						self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*a_   )



							def       _a   (      self   )							->							Optional[int]:
								_UpperCAmelCase =						self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_image_classification(*a_   )



							def       _a   (      self   )							->							Dict:
								_UpperCAmelCase =						self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_backbone(*a_   )



							@unittest.skip(reason="Dinat does not use inputs_embeds"   )
							def       _a   (      self   )							->							str:
								pass



							@unittest.skip(reason="Dinat does not use feedforward chunking"   )
							def       _a   (      self   )							->							List[Any]:
								pass



							def       _a   (      self   )							->							List[Any]:
								_UpperCAmelCase ,      _UpperCAmelCase =						self.model_tester.prepare_config_and_inputs_for_common()

								for model_class in self.all_model_classes:
									_UpperCAmelCase =						model_class(a_   )
									self.assertIsInstance(model.get_input_embeddings() ,						(nn.Module)   )
									_UpperCAmelCase =						model.get_output_embeddings()
									self.assertTrue(x is None or isinstance(a_ ,						nn.Linear   )   )



							def       _a   (      self   )							->							Dict:
								_UpperCAmelCase ,      _UpperCAmelCase =						self.model_tester.prepare_config_and_inputs_for_common()

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

									_UpperCAmelCase =						["pixel_values"]
									self.assertListEqual(arg_names[:1] ,						a_   )



							def       _a   (      self   )							->							Tuple:
								self.skipTest("Dinat's attention operation is handled entirely by NATTEN."   )



							def       _a   (      self ,						a_ ,						a_ ,						a_ ,						a_   )							->							int:
								_UpperCAmelCase =						model_class(a_   )
								model.to(a_   )
								model.eval()

								with torch.no_grad():
									_UpperCAmelCase =						model(**self._prepare_for_class(a_ ,						a_   )   )

								_UpperCAmelCase =						outputs.hidden_states

								_UpperCAmelCase =						getattr(
								    self.model_tester ,						"expected_num_hidden_layers" ,						len(self.model_tester.depths   ) + 1   )
								self.assertEqual(len(a_   ) ,						a_   )

								# Dinat has a different seq_length
								_UpperCAmelCase =						(
								    config.patch_size
								    if isinstance(config.patch_size ,						collections.abc.Iterable   )
								    else (config.patch_size, config.patch_size)
								)

								_UpperCAmelCase =						image_size[0] // patch_size[0]
								_UpperCAmelCase =						image_size[1] // patch_size[1]

								self.assertListEqual(
								    list(hidden_states[0].shape[-3:]   ) ,						[height, width, self.model_tester.embed_dim] ,						)

								if model_class.__name__ != "DinatBackbone":
									_UpperCAmelCase =						outputs.reshaped_hidden_states
									self.assertEqual(len(a_   ) ,						a_   )

									_UpperCAmelCase ,      _UpperCAmelCase ,      _UpperCAmelCase ,      _UpperCAmelCase =						reshaped_hidden_states[0].shape
									_UpperCAmelCase =						(
									    reshaped_hidden_states[0].view(a_ ,						a_ ,						a_ ,						a_   ).permute(0 ,						2 ,						3 ,						1   )
									)
									self.assertListEqual(
									    list(reshaped_hidden_states.shape[-3:]   ) ,						[height, width, self.model_tester.embed_dim] ,						)



							def       _a   (      self   )							->							List[str]:
								_UpperCAmelCase ,      _UpperCAmelCase =						self.model_tester.prepare_config_and_inputs_for_common()

								_UpperCAmelCase =						(
								    self.model_tester.image_size
								    if isinstance(self.model_tester.image_size ,						collections.abc.Iterable   )
								    else (self.model_tester.image_size, self.model_tester.image_size)
								)

								for model_class in self.all_model_classes:
									_UpperCAmelCase =						True
									self.check_hidden_states_output(a_ ,						a_ ,						a_ ,						a_   )

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

									self.check_hidden_states_output(a_ ,						a_ ,						a_ ,						a_   )



							@slow
							def       _a   (      self   )							->							int:
								for model_name in DINAT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
									_UpperCAmelCase =						DinatModel.from_pretrained(a_   )
									self.assertIsNotNone(a_   )



							def       _a   (      self   )							->							List[Any]:
								_UpperCAmelCase ,      _UpperCAmelCase =						self.model_tester.prepare_config_and_inputs_for_common()

								_UpperCAmelCase =						_config_zero_init(a_   )
								for model_class in self.all_model_classes:
									_UpperCAmelCase =						model_class(config=a_   )
									for name, param in model.named_parameters():
										if "embeddings" not in name and param.requires_grad:
											self.assertIn(
											    ((param.data.mean() * 1e9).round() / 1e9).item() ,						[0.0, 1.0] ,						msg=f"Parameter {name} of model {model_class} seems not properly initialized" ,						)





@require_natten
@require_vision
@require_torch
class    _lowerCAmelCase						(						unittest.TestCase						):



							@cached_property
							def       _a   (      self   )							->							Optional[Any]:
								return AutoImageProcessor.from_pretrained("shi-labs/dinat-mini-in1k-224"   ) if is_vision_available() else None



							@slow
							def       _a   (      self   )							->							Optional[Any]:
								_UpperCAmelCase =						DinatForImageClassification.from_pretrained("shi-labs/dinat-mini-in1k-224"   ).to(a_   )
								_UpperCAmelCase =						self.default_image_processor

								_UpperCAmelCase =						Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"   )
								_UpperCAmelCase =						image_processor(images=a_ ,						return_tensors="pt"   ).to(a_   )

								# forward pass
								with torch.no_grad():
									_UpperCAmelCase =						model(**a_   )

								# verify the logits
								_UpperCAmelCase =						torch.Size((1, 1000)   )
								self.assertEqual(outputs.logits.shape ,						a_   )
								_UpperCAmelCase =						torch.tensor([-0.1545, -0.7667, 0.4642]   ).to(a_   )
								self.assertTrue(torch.allclose(outputs.logits[0, :3] ,						a_ ,						atol=1e-4   )   )





@require_torch
@require_natten
class    _lowerCAmelCase						(						unittest.TestCase    ,       lowerCamelCase						):
							lowercase_     :     Dict       =							(DinatBackbone,) if is_torch_available() else ()
							lowercase_     :     List[str]       =							DinatConfig



							def       _a   (      self   )							->							List[str]:
								_UpperCAmelCase =						DinatModelTester(self   )






"""simple docstring"""
import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
			from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

__magic_name__		       =				logging.get_logger(__name__)




def     __lowerCamelCase  (		UpperCamelCase__				,     UpperCamelCase__      ):


	"""simple docstring"""






	_UpperCAmelCase =						f"{sampling_rate}"
	_UpperCAmelCase =						"1"
	_UpperCAmelCase =						"f32le"
	_UpperCAmelCase =						[
	    "ffmpeg",
	    "-i",
	    "pipe:0",
	    "-ac",
	    ac,
	    "-ar",
	    ar,
	    "-f",
	    format_for_conversion,
	    "-hide_banner",
	    "-loglevel",
	    "quiet",
	    "pipe:1",
	]

	try:
		_UpperCAmelCase =						subprocess.Popen(UpperCamelCase__				,     stdin=subprocess.PIPE				,     stdout=subprocess.PIPE      )
	except FileNotFoundError:
		raise ValueError("ffmpeg was not found but is required to load audio files from filename"      )
	_UpperCAmelCase =						ffmpeg_process.communicate(UpperCamelCase__      )
	_UpperCAmelCase =						output_stream[0]

	_UpperCAmelCase =						np.frombuffer(UpperCamelCase__				,     np.floataa      )
	if audio.shape[0] == 0:
		raise ValueError("Malformed soundfile"      )
	return audio



@add_end_docstrings(lowerCamelCase						)
class    _lowerCAmelCase						(						lowerCamelCase						):



							def __init__(      self ,						*a_ ,						**a_   )							->							Union[str, Any]:
								# Default, might be overriden by the model.config.
								_UpperCAmelCase =						5
								super().__init__(*a_ ,						**a_   )

								if self.framework != "pt":
									raise ValueError(f"The {self.__class__} is only available in PyTorch."   )

								self.check_model_type(a_   )



							def __call__(      self ,						a_ ,						**a_ ,						)							->							int:
								return super().__call__(a_ ,						**a_   )



							def       _a   (      self ,						a_=None ,						**a_   )							->							Dict:
								# No parameters on this pipeline right now
								_UpperCAmelCase =						{}
								if top_k is not None:
									if top_k > self.model.config.num_labels:
										_UpperCAmelCase =						self.model.config.num_labels
									_UpperCAmelCase =						top_k
								return {}, {}, postprocess_params



							def       _a   (      self ,						a_   )							->							str:
								if isinstance(a_ ,						a_   ):
									if inputs.startswith("http://"   ) or inputs.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
										_UpperCAmelCase =						requests.get(a_   ).content
									else:
										with open(a_ ,						"rb"   ) as f:
											_UpperCAmelCase =						f.read()

								if isinstance(a_ ,						a_   ):
									_UpperCAmelCase =						ffmpeg_read(a_ ,						self.feature_extractor.sampling_rate   )

								if isinstance(a_ ,						a_   ):
									# Accepting `"array"` which is the key defined in `datasets` for
									# better integration
									if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
										raise ValueError(
										    "When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a "
										    "\"raw\" key containing the numpy array representing the audio and a \"sampling_rate\" key, "
										    "containing the sampling_rate associated with that array"   )

									_UpperCAmelCase =						inputs.pop("raw" ,						a_   )
									if _inputs is None:
										# Remove path which will not be used from `datasets`.
										inputs.pop("path" ,						a_   )
										_UpperCAmelCase =						inputs.pop("array" ,						a_   )
									_UpperCAmelCase =						inputs.pop("sampling_rate"   )
									_UpperCAmelCase =						_inputs
									if in_sampling_rate != self.feature_extractor.sampling_rate:
										import torch

										if is_torchaudio_available():
											from torchaudio import functional as F
										else:
											raise ImportError(
											    "torchaudio is required to resample audio samples in AudioClassificationPipeline. "
											    "The torchaudio package can be installed through: `pip install torchaudio`."   )

										_UpperCAmelCase =						F.resample(
										    torch.from_numpy(a_   ) ,						a_ ,						self.feature_extractor.sampling_rate   ).numpy()

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

								_UpperCAmelCase =						self.feature_extractor(
								    a_ ,						sampling_rate=self.feature_extractor.sampling_rate ,						return_tensors="pt"   )
								return processed



							def       _a   (      self ,						a_   )							->							List[str]:
								_UpperCAmelCase =						self.model(**a_   )
								return model_outputs



							def       _a   (      self ,						a_ ,						a_=5   )							->							int:
								_UpperCAmelCase =						model_outputs.logits[0].softmax(-1   )
								_UpperCAmelCase ,      _UpperCAmelCase =						probs.topk(a_   )

								_UpperCAmelCase =						scores.tolist()
								_UpperCAmelCase =						ids.tolist()

								_UpperCAmelCase =						[{"score": score, "label": self.model.config.idalabel[_id]} for score, _id in zip(a_ ,						a_   )]

								return labels




