

import inspect
import unittest

from transformers import is_torch_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device

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 MODEL_MAPPING, GLPNConfig, GLPNForDepthEstimation, GLPNModel
		from transformers.models.glpn.modeling_glpn import GLPN_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
		from PIL import Image

		from transformers import GLPNImageProcessor





class      snake_case_      (					_a				):


							"""simple docstring"""





							def 							A__	(			self				):
												__lowerCAmelCase       =				self.config_class(**self.inputs_dict				)
												self.parent.assertTrue(hasattr(_A       ,    'hidden_sizes'				)				)
												self.parent.assertTrue(hasattr(_A       ,    'num_attention_heads'				)				)
												self.parent.assertTrue(hasattr(_A       ,    'num_encoder_blocks'				)				)





class      snake_case_      :


							"""simple docstring"""





							def __init__(			self       ,    _A       ,    _A=1_3       ,    _A=6_4       ,    _A=3       ,    _A=4       ,    _A=[2, 2, 2, 2]       ,    _A=[8, 4, 2, 1]       ,    _A=[1_6, 3_2, 6_4, 1_2_8]       ,    _A=[1, 4, 8, 1_6]       ,    _A=[1, 2, 4, 8]       ,    _A=True       ,    _A=True       ,    _A="gelu"       ,    _A=0.1       ,    _A=0.1       ,    _A=0.02       ,    _A=1_6       ,    _A=3       ,    _A=None       ,    ):
												__lowerCAmelCase       =				parent
												__lowerCAmelCase       =				batch_size
												__lowerCAmelCase       =				image_size
												__lowerCAmelCase       =				num_channels
												__lowerCAmelCase       =				num_encoder_blocks
												__lowerCAmelCase       =				sr_ratios
												__lowerCAmelCase       =				depths
												__lowerCAmelCase       =				hidden_sizes
												__lowerCAmelCase       =				downsampling_rates
												__lowerCAmelCase       =				num_attention_heads
												__lowerCAmelCase       =				is_training
												__lowerCAmelCase       =				use_labels
												__lowerCAmelCase       =				hidden_act
												__lowerCAmelCase       =				hidden_dropout_prob
												__lowerCAmelCase       =				attention_probs_dropout_prob
												__lowerCAmelCase       =				initializer_range
												__lowerCAmelCase       =				decoder_hidden_size
												__lowerCAmelCase       =				num_labels
												__lowerCAmelCase       =				scope




							def 							A__	(			self				):
												__lowerCAmelCase       =				floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size]				)

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

												__lowerCAmelCase       =				self.get_config()
												return config, pixel_values, labels




							def 							A__	(			self				):
												return GLPNConfig(
												    image_size=self.image_size       ,    num_channels=self.num_channels       ,    num_encoder_blocks=self.num_encoder_blocks       ,    depths=self.depths       ,    hidden_sizes=self.hidden_sizes       ,    num_attention_heads=self.num_attention_heads       ,    hidden_act=self.hidden_act       ,    hidden_dropout_prob=self.hidden_dropout_prob       ,    attention_probs_dropout_prob=self.attention_probs_dropout_prob       ,    initializer_range=self.initializer_range       ,    decoder_hidden_size=self.decoder_hidden_size       ,    )




							def 							A__	(			self       ,    _A       ,    _A       ,    _A				):
												__lowerCAmelCase       =				GLPNModel(config=_A				)
												model.to(_A				)
												model.eval()
												__lowerCAmelCase       =				model(_A				)
												__lowerCAmelCase       =				__lowerCAmelCase       =				self.image_size // (self.downsampling_rates[-1] * 2)
												self.parent.assertEqual(
												    result.last_hidden_state.shape       ,    (self.batch_size, self.hidden_sizes[-1], expected_height, expected_width)				)




							def 							A__	(			self       ,    _A       ,    _A       ,    _A				):
												__lowerCAmelCase       =				self.num_labels
												__lowerCAmelCase       =				GLPNForDepthEstimation(_A				)
												model.to(_A				)
												model.eval()
												__lowerCAmelCase       =				model(_A				)
												self.parent.assertEqual(result.predicted_depth.shape       ,    (self.batch_size, self.image_size, self.image_size)				)
												__lowerCAmelCase       =				model(_A       ,    labels=_A				)
												self.parent.assertEqual(result.predicted_depth.shape       ,    (self.batch_size, self.image_size, self.image_size)				)




							def 							A__	(			self				):
												__lowerCAmelCase       =				self.prepare_config_and_inputs()
												__lowerCAmelCase,						__lowerCAmelCase,						__lowerCAmelCase       =				config_and_inputs
												__lowerCAmelCase       =				{'pixel_values': pixel_values}
												return config, inputs_dict





@require_torch
class      snake_case_      (					_a      ,			_a      ,			unittest.TestCase				):


							"""simple docstring"""





							__UpperCAmelCase      =(GLPNModel, GLPNForDepthEstimation) if is_torch_available() else ()
							__UpperCAmelCase      =(
							    {"""depth-estimation""": GLPNForDepthEstimation, """feature-extraction""": GLPNModel} if is_torch_available() else {}
							)

							__UpperCAmelCase      =False
							__UpperCAmelCase      =False
							__UpperCAmelCase      =False




							def 							A__	(			self				):
												__lowerCAmelCase       =				GLPNModelTester(self				)
												__lowerCAmelCase       =				GLPNConfigTester(self       ,    config_class=_A				)




							def 							A__	(			self				):
												self.config_tester.run_common_tests()




							def 							A__	(			self				):
												__lowerCAmelCase       =				self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_model(*_A				)




							def 							A__	(			self				):
												__lowerCAmelCase       =				self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_for_depth_estimation(*_A				)




							@unittest.skip('GLPN does not use inputs_embeds'				)
							def 							A__	(			self				):
												pass




							@unittest.skip('GLPN does not have get_input_embeddings method and get_output_embeddings methods'				)
							def 							A__	(			self				):
												pass




							def 							A__	(			self				):
												__lowerCAmelCase,						__lowerCAmelCase       =				self.model_tester.prepare_config_and_inputs_for_common()

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

																	__lowerCAmelCase       =				['pixel_values']
																	self.assertListEqual(arg_names[:1]       ,    _A				)




							def 							A__	(			self				):
												__lowerCAmelCase,						__lowerCAmelCase       =				self.model_tester.prepare_config_and_inputs_for_common()
												__lowerCAmelCase       =				True

												for model_class in self.all_model_classes:
																	__lowerCAmelCase       =				True
																	__lowerCAmelCase       =				False
																	__lowerCAmelCase       =				True
																	__lowerCAmelCase       =				model_class(_A				)
																	model.to(_A				)
																	model.eval()
																	with torch.no_grad():
																						__lowerCAmelCase       =				model(**self._prepare_for_class(_A       ,    _A				)				)
																	__lowerCAmelCase       =				outputs.attentions

																	__lowerCAmelCase       =				sum(self.model_tester.depths				)
																	self.assertEqual(len(_A				)       ,    _A				)

																	# check that output_attentions also work using config
																	del inputs_dict["output_attentions"]
																	__lowerCAmelCase       =				True
																	__lowerCAmelCase       =				model_class(_A				)
																	model.to(_A				)
																	model.eval()
																	with torch.no_grad():
																						__lowerCAmelCase       =				model(**self._prepare_for_class(_A       ,    _A				)				)
																	__lowerCAmelCase       =				outputs.attentions

																	self.assertEqual(len(_A				)       ,    _A				)

																	# verify the first attentions (first block, first layer)
																	__lowerCAmelCase       =				(self.model_tester.image_size // 4) ** 2
																	__lowerCAmelCase       =				(self.model_tester.image_size // (4 * self.model_tester.sr_ratios[0])) ** 2
																	self.assertListEqual(
																	    list(attentions[0].shape[-3:]				)       ,    [self.model_tester.num_attention_heads[0], expected_seq_len, expected_reduced_seq_len]       ,    )

																	# verify the last attentions (last block, last layer)
																	__lowerCAmelCase       =				(self.model_tester.image_size // 3_2) ** 2
																	__lowerCAmelCase       =				(self.model_tester.image_size // (3_2 * self.model_tester.sr_ratios[-1])) ** 2
																	self.assertListEqual(
																	    list(attentions[-1].shape[-3:]				)       ,    [self.model_tester.num_attention_heads[-1], expected_seq_len, expected_reduced_seq_len]       ,    )
																	__lowerCAmelCase       =				len(_A				)

																	# Check attention is always last and order is fine
																	__lowerCAmelCase       =				True
																	__lowerCAmelCase       =				True
																	__lowerCAmelCase       =				model_class(_A				)
																	model.to(_A				)
																	model.eval()
																	with torch.no_grad():
																						__lowerCAmelCase       =				model(**self._prepare_for_class(_A       ,    _A				)				)

																	self.assertEqual(out_len + 1       ,    len(_A				)				)

																	__lowerCAmelCase       =				outputs.attentions

																	self.assertEqual(len(_A				)       ,    _A				)
																	# verify the first attentions (first block, first layer)
																	__lowerCAmelCase       =				(self.model_tester.image_size // 4) ** 2
																	__lowerCAmelCase       =				(self.model_tester.image_size // (4 * self.model_tester.sr_ratios[0])) ** 2
																	self.assertListEqual(
																	    list(self_attentions[0].shape[-3:]				)       ,    [self.model_tester.num_attention_heads[0], expected_seq_len, expected_reduced_seq_len]       ,    )




							def 							A__	(			self				):
												def check_hidden_states_output(_A       ,    _A       ,    _A				):
																	__lowerCAmelCase       =				model_class(_A				)
																	model.to(_A				)
																	model.eval()

																	with torch.no_grad():
																						__lowerCAmelCase       =				model(**self._prepare_for_class(_A       ,    _A				)				)

																	__lowerCAmelCase       =				outputs.hidden_states

																	__lowerCAmelCase       =				self.model_tester.num_encoder_blocks
																	self.assertEqual(len(_A				)       ,    _A				)

																	# verify the first hidden states (first block)
																	self.assertListEqual(
																	    list(hidden_states[0].shape[-3:]				)       ,    [
																	        self.model_tester.hidden_sizes[0],
																	        self.model_tester.image_size // 4,
																	        self.model_tester.image_size // 4,
																	    ]       ,    )

												__lowerCAmelCase,						__lowerCAmelCase       =				self.model_tester.prepare_config_and_inputs_for_common()

												for model_class in self.all_model_classes:
																	__lowerCAmelCase       =				True
																	check_hidden_states_output(_A       ,    _A       ,    _A				)

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

																	check_hidden_states_output(_A       ,    _A       ,    _A				)




							def 							A__	(			self				):
												if not self.model_tester.is_training:
																	return

												__lowerCAmelCase,						__lowerCAmelCase       =				self.model_tester.prepare_config_and_inputs_for_common()
												__lowerCAmelCase       =				True

												for model_class in self.all_model_classes:
																	if model_class in get_values(_A				):
																						continue
																	# TODO: remove the following 3 lines once we have a MODEL_FOR_DEPTH_ESTIMATION_MAPPING
																	# this can then be incorporated into _prepare_for_class in test_modeling_common.py
																	if model_class.__name__ == "GLPNForDepthEstimation":
																						__lowerCAmelCase,						__lowerCAmelCase,						__lowerCAmelCase,						__lowerCAmelCase       =				inputs_dict['pixel_values'].shape
																						__lowerCAmelCase       =				torch.zeros(
																						    [self.model_tester.batch_size, height, width]       ,    device=_A				).long()
																	__lowerCAmelCase       =				model_class(_A				)
																	model.to(_A				)
																	model.train()
																	__lowerCAmelCase       =				self._prepare_for_class(_A       ,    _A       ,    return_labels=_A				)
																	__lowerCAmelCase       =				model(**_A				).loss
																	loss.backward()




							@slow
							def 							A__	(			self				):
												for model_name in GLPN_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	__lowerCAmelCase       =				GLPNModel.from_pretrained(_A				)
																	self.assertIsNotNone(_A				)


def 							__lowercase     (	):

					"""simple docstring"""






					__lowerCAmelCase       =				Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png'   )
					return image





@require_torch
@require_vision
@slow
class      snake_case_      (					unittest.TestCase				):


							"""simple docstring"""





							@slow
							def 							A__	(			self				):
												__lowerCAmelCase       =				GLPNImageProcessor.from_pretrained(GLPN_PRETRAINED_MODEL_ARCHIVE_LIST[0]				)
												__lowerCAmelCase       =				GLPNForDepthEstimation.from_pretrained(GLPN_PRETRAINED_MODEL_ARCHIVE_LIST[0]				).to(_A				)

												__lowerCAmelCase       =				prepare_img()
												__lowerCAmelCase       =				image_processor(images=_A       ,    return_tensors='pt'				).to(_A				)

												# forward pass
												with torch.no_grad():
																	__lowerCAmelCase       =				model(**_A				)

												# verify the predicted depth
												__lowerCAmelCase       =				torch.Size([1, 4_8_0, 6_4_0]				)
												self.assertEqual(outputs.predicted_depth.shape       ,    _A				)

												__lowerCAmelCase       =				torch.tensor(
												    [[3.4291, 2.7865, 2.5151], [3.2841, 2.7021, 2.3502], [3.1147, 2.4625, 2.2481]]				).to(_A				)

												self.assertTrue(torch.allclose(outputs.predicted_depth[0, :3, :3]       ,    _A       ,    atol=1e-4				)				)









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

lowerCamelCase   					=  logging.get_logger(__name__)


def 							__lowercase     (	UpperCAmelCase__		,       UpperCAmelCase__   ):

					"""simple docstring"""






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

					try:
										__lowerCAmelCase       =				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'   )
					__lowerCAmelCase       =				ffmpeg_process.communicate(UpperCAmelCase__   )
					__lowerCAmelCase       =				output_stream[0]

					__lowerCAmelCase       =				np.frombuffer(UpperCAmelCase__		,       np.floataa   )
					if audio.shape[0] == 0:
										raise ValueError('Malformed soundfile'   )
					return audio





@add_end_docstrings(_a				)
class      snake_case_      (					_a				):


							"""simple docstring"""





							def __init__(			self       ,    *_A       ,    **_A				):
												# Default, might be overriden by the model.config.
												__lowerCAmelCase       =				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       ,    ):
												return super().__call__(_A       ,    **_A				)




							def 							A__	(			self       ,    _A=None       ,    **_A				):
												# No parameters on this pipeline right now
												__lowerCAmelCase       =				{}
												if top_k is not None:
																	if top_k > self.model.config.num_labels:
																						__lowerCAmelCase       =				self.model.config.num_labels
																	__lowerCAmelCase       =				top_k
												return {}, {}, postprocess_params




							def 							A__	(			self       ,    _A				):
												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
																						__lowerCAmelCase       =				requests.get(_A				).content
																	else:
																						with open(_A       ,    'rb'				) as f:
																											__lowerCAmelCase       =				f.read()

												if isinstance(_A       ,    _A				):
																	__lowerCAmelCase       =				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'				)

																	__lowerCAmelCase       =				inputs.pop('raw'       ,    _A				)
																	if _inputs is None:
																						# Remove path which will not be used from `datasets`.
																						inputs.pop('path'       ,    _A				)
																						__lowerCAmelCase       =				inputs.pop('array'       ,    _A				)
																	__lowerCAmelCase       =				inputs.pop('sampling_rate'				)
																	__lowerCAmelCase       =				_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`.'				)

																						__lowerCAmelCase       =				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'				)

												__lowerCAmelCase       =				self.feature_extractor(
												    _A       ,    sampling_rate=self.feature_extractor.sampling_rate       ,    return_tensors='pt'				)
												return processed




							def 							A__	(			self       ,    _A				):
												__lowerCAmelCase       =				self.model(**_A				)
												return model_outputs




							def 							A__	(			self       ,    _A       ,    _A=5				):
												__lowerCAmelCase       =				model_outputs.logits[0].softmax(-1				)
												__lowerCAmelCase,						__lowerCAmelCase       =				probs.topk(_A				)

												__lowerCAmelCase       =				scores.tolist()
												__lowerCAmelCase       =				ids.tolist()

												__lowerCAmelCase       =				[{'score': score, 'label': self.model.config.idalabel[_id]} for score, _id in zip(_A       ,    _A				)]

												return labels






