






"""simple docstring"""


import itertools
import random
import unittest

import numpy as np

from transformers import WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST, WavaVecaConfig, WavaVecaFeatureExtractor
from transformers.testing_utils import require_torch, slow

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


__SCREAMING_SNAKE_CASE   =   random.Random()


def        UpperCAmelCase      (					a__     ,      a__=1.0     ,      a__=None     ,      a__=None ):





						'''simple docstring'''






						if rng is None:
												lowerCAmelCase   :int						=			global_rng

						lowerCAmelCase   :str						=			[]
						for batch_idx in range(shape[0] ):
												values.append([] )
												for _ in range(shape[1] ):
																		values[-1].append(rng.random() * scale )

						return values




class 						__UpperCamelCase     (     unittest.TestCase						):
					def __init__(							self							:	str     , UpperCAmelCase							:	List[str]     , UpperCAmelCase							:	Tuple=7     , UpperCAmelCase							:	Any=400     , UpperCAmelCase							:	Optional[int]=2000     , UpperCAmelCase							:	Optional[Any]=1     , UpperCAmelCase							:	Tuple=0.0     , UpperCAmelCase							:	str=1_6000     , UpperCAmelCase							:	List[str]=True     , UpperCAmelCase							:	Dict=True     , )	->       int:
											lowerCAmelCase   :List[str]						=			parent
											lowerCAmelCase   :Tuple						=			batch_size
											lowerCAmelCase   :int						=			min_seq_length
											lowerCAmelCase   :Optional[Any]						=			max_seq_length
											lowerCAmelCase   :Dict						=			(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
											lowerCAmelCase   :str						=			feature_size
											lowerCAmelCase   :Tuple						=			padding_value
											lowerCAmelCase   :Union[str, Any]						=			sampling_rate
											lowerCAmelCase   :Tuple						=			return_attention_mask
											lowerCAmelCase   :int						=			do_normalize
					def        UpperCAmelCase__ (							self							:	Optional[int]   )	->       Union[str, Any]:
											return {
											    "feature_size": self.feature_size,
											    "padding_value": self.padding_value,
											    "sampling_rate": self.sampling_rate,
											    "return_attention_mask": self.return_attention_mask,
											    "do_normalize": self.do_normalize,
											}
					def        UpperCAmelCase__ (							self							:	Tuple     , UpperCAmelCase							:	Tuple=False     , UpperCAmelCase							:	Optional[Any]=False   )	->       Tuple:
											def _flatten(UpperCAmelCase							:	Tuple   ):
																	return list(itertools.chain(*UpperCAmelCase   )   )

											if equal_length:
																	lowerCAmelCase   :Any						=			floats_list((self.batch_size, self.max_seq_length)   )
											else:
																	# make sure that inputs increase in size
																	lowerCAmelCase   :List[str]						=			[
																	    _flatten(floats_list((x, self.feature_size)   )   )
																	    for x in range(self.min_seq_length     , self.max_seq_length     , self.seq_length_diff   )
																	]

											if numpify:
																	lowerCAmelCase   :List[str]						=			[np.asarray(UpperCAmelCase   ) for x in speech_inputs]

											return speech_inputs




class 						__UpperCamelCase     (     UpperCamelCase   ,     unittest.TestCase						):
					lowercase_    :				Dict      =  WavaVecaFeatureExtractor
					def        UpperCAmelCase__ (							self							:	List[str]   )	->       Optional[int]:
											lowerCAmelCase   :List[str]						=			WavaVecaFeatureExtractionTester(self   )
					def        UpperCAmelCase__ (							self							:	Optional[Any]     , UpperCAmelCase							:	str   )	->       str:
											self.assertTrue(np.all(np.mean(UpperCAmelCase     , axis=0   ) < 1e-3   )   )
											self.assertTrue(np.all(np.abs(np.var(UpperCAmelCase     , axis=0   ) - 1   ) < 1e-3   )   )
					def        UpperCAmelCase__ (							self							:	int   )	->       List[Any]:
											# Tests that all call wrap to encode_plus and batch_encode_plus
											lowerCAmelCase   :List[str]						=			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
											# create three inputs of length 800, 1000, and 1200
											lowerCAmelCase   :Optional[int]						=			[floats_list((1, x)   )[0] for x in range(800     , 1400     , 200   )]
											lowerCAmelCase   :int						=			[np.asarray(UpperCAmelCase   ) for speech_input in speech_inputs]

											# Test not batched input
											lowerCAmelCase   :Union[str, Any]						=			feat_extract(speech_inputs[0]     , return_tensors='np'   ).input_values
											lowerCAmelCase   :List[str]						=			feat_extract(np_speech_inputs[0]     , return_tensors='np'   ).input_values
											self.assertTrue(np.allclose(UpperCAmelCase     , UpperCAmelCase     , atol=1e-3   )   )

											# Test batched
											lowerCAmelCase   :Dict						=			feat_extract(UpperCAmelCase     , return_tensors='np'   ).input_values
											lowerCAmelCase   :Tuple						=			feat_extract(UpperCAmelCase     , return_tensors='np'   ).input_values
											for enc_seq_a, enc_seq_a in zip(UpperCAmelCase     , UpperCAmelCase   ):
																	self.assertTrue(np.allclose(UpperCAmelCase     , UpperCAmelCase     , atol=1e-3   )   )

											# Test 2-D numpy arrays are batched.
											lowerCAmelCase   :Tuple						=			[floats_list((1, x)   )[0] for x in (800, 800, 800)]
											lowerCAmelCase   :List[Any]						=			np.asarray(UpperCAmelCase   )
											lowerCAmelCase   :Any						=			feat_extract(UpperCAmelCase     , return_tensors='np'   ).input_values
											lowerCAmelCase   :Optional[int]						=			feat_extract(UpperCAmelCase     , return_tensors='np'   ).input_values
											for enc_seq_a, enc_seq_a in zip(UpperCAmelCase     , UpperCAmelCase   ):
																	self.assertTrue(np.allclose(UpperCAmelCase     , UpperCAmelCase     , atol=1e-3   )   )
					def        UpperCAmelCase__ (							self							:	int   )	->       Optional[Any]:
											lowerCAmelCase   :int						=			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
											lowerCAmelCase   :int						=			[floats_list((1, x)   )[0] for x in range(800     , 1400     , 200   )]

											lowerCAmelCase   :List[str]						=			['longest', 'max_length', 'do_not_pad']
											lowerCAmelCase   :str						=			[None, 1600, None]
											for max_length, padding in zip(UpperCAmelCase     , UpperCAmelCase   ):
																	lowerCAmelCase   :Union[str, Any]						=			feat_extract(UpperCAmelCase     , padding=UpperCAmelCase     , max_length=UpperCAmelCase     , return_tensors='np'   )
																	lowerCAmelCase   :Union[str, Any]						=			processed.input_values

																	self._check_zero_mean_unit_variance(input_values[0][:800]   )
																	self.assertTrue(input_values[0][800:].sum() < 1e-6   )
																	self._check_zero_mean_unit_variance(input_values[1][:1000]   )
																	self.assertTrue(input_values[0][1000:].sum() < 1e-6   )
																	self._check_zero_mean_unit_variance(input_values[2][:1200]   )
					def        UpperCAmelCase__ (							self							:	Optional[Any]   )	->       Optional[Any]:
											lowerCAmelCase   :Optional[int]						=			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
											lowerCAmelCase   :Union[str, Any]						=			range(800     , 1400     , 200   )
											lowerCAmelCase   :Any						=			[floats_list((1, x)   )[0] for x in lengths]

											lowerCAmelCase   :Tuple						=			['longest', 'max_length', 'do_not_pad']
											lowerCAmelCase   :Optional[int]						=			[None, 1600, None]

											for max_length, padding in zip(UpperCAmelCase     , UpperCAmelCase   ):
																	lowerCAmelCase   :Dict						=			feat_extract(UpperCAmelCase     , max_length=UpperCAmelCase     , padding=UpperCAmelCase   )
																	lowerCAmelCase   :Optional[Any]						=			processed.input_values

																	self._check_zero_mean_unit_variance(input_values[0][:800]   )
																	self._check_zero_mean_unit_variance(input_values[1][:1000]   )
																	self._check_zero_mean_unit_variance(input_values[2][:1200]   )
					def        UpperCAmelCase__ (							self							:	Any   )	->       List[Any]:
											lowerCAmelCase   :str						=			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
											lowerCAmelCase   :Tuple						=			[floats_list((1, x)   )[0] for x in range(800     , 1400     , 200   )]
											lowerCAmelCase   :List[Any]						=			feat_extract(
											    UpperCAmelCase     , truncation=UpperCAmelCase     , max_length=1000     , padding='max_length'     , return_tensors='np'   )
											lowerCAmelCase   :str						=			processed.input_values

											self._check_zero_mean_unit_variance(input_values[0, :800]   )
											self._check_zero_mean_unit_variance(input_values[1]   )
											self._check_zero_mean_unit_variance(input_values[2]   )
					def        UpperCAmelCase__ (							self							:	List[Any]   )	->       List[Any]:
											lowerCAmelCase   :Union[str, Any]						=			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
											lowerCAmelCase   :Union[str, Any]						=			[floats_list((1, x)   )[0] for x in range(800     , 1400     , 200   )]
											lowerCAmelCase   :int						=			feat_extract(
											    UpperCAmelCase     , truncation=UpperCAmelCase     , max_length=1000     , padding='longest'     , return_tensors='np'   )
											lowerCAmelCase   :List[Any]						=			processed.input_values

											self._check_zero_mean_unit_variance(input_values[0, :800]   )
											self._check_zero_mean_unit_variance(input_values[1, :1000]   )
											self._check_zero_mean_unit_variance(input_values[2]   )

											# make sure that if max_length < longest -> then pad to max_length
											self.assertTrue(input_values.shape == (3, 1000)   )

											lowerCAmelCase   :Optional[int]						=			[floats_list((1, x)   )[0] for x in range(800     , 1400     , 200   )]
											lowerCAmelCase   :Tuple						=			feat_extract(
											    UpperCAmelCase     , truncation=UpperCAmelCase     , max_length=2000     , padding='longest'     , return_tensors='np'   )
											lowerCAmelCase   :str						=			processed.input_values

											self._check_zero_mean_unit_variance(input_values[0, :800]   )
											self._check_zero_mean_unit_variance(input_values[1, :1000]   )
											self._check_zero_mean_unit_variance(input_values[2]   )

											# make sure that if max_length > longest -> then pad to longest
											self.assertTrue(input_values.shape == (3, 1200)   )
					@require_torch
					def        UpperCAmelCase__ (							self							:	Optional[Any]   )	->       Optional[Any]:
											import torch

											lowerCAmelCase   :Union[str, Any]						=			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()   )
											lowerCAmelCase   :int						=			np.random.rand(100   ).astype(np.floataa   )
											lowerCAmelCase   :Any						=			np_speech_inputs.tolist()

											for inputs in [py_speech_inputs, np_speech_inputs]:
																	lowerCAmelCase   :Tuple						=			feature_extractor.pad([{'input_values': inputs}]     , return_tensors='np'   )
																	self.assertTrue(np_processed.input_values.dtype == np.floataa   )
																	lowerCAmelCase   :Union[str, Any]						=			feature_extractor.pad([{'input_values': inputs}]     , return_tensors='pt'   )
																	self.assertTrue(pt_processed.input_values.dtype == torch.floataa   )
					@slow
					@require_torch
					def        UpperCAmelCase__ (							self							:	Dict   )	->       Optional[Any]:
											# this test makes sure that models that are using
											# group norm don't have their feature extractor return the
											# attention_mask
											for model_id in WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST:
																	lowerCAmelCase   :str						=			WavaVecaConfig.from_pretrained(UpperCAmelCase   )
																	lowerCAmelCase   :List[Any]						=			WavaVecaFeatureExtractor.from_pretrained(UpperCAmelCase   )

																	# only "layer" feature extraction norm should make use of
																	# attention_mask
																	self.assertEqual(feat_extract.return_attention_mask     , config.feat_extract_norm == 'layer'   )







"""simple docstring"""


from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
		from PIL import Image

		from ..image_utils import load_image

if is_torch_available():
		import torch

		from transformers.modeling_outputs import BaseModelOutput

		from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

__SCREAMING_SNAKE_CASE   =   logging.get_logger(__name__)




@add_end_docstrings(UpperCamelCase						)
class 						__UpperCamelCase     (     UpperCamelCase						):
					def __init__(							self							:	Union[str, Any]     , **UpperCAmelCase							:	List[Any]   )	->       int:
											super().__init__(**UpperCAmelCase   )

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

											requires_backends(self     , 'vision'   )
											self.check_model_type(UpperCAmelCase   )
					def __call__(							self							:	Optional[Any]     , UpperCAmelCase							:	Union[str, "Image.Image", List[Dict[str, Any]]]     , UpperCAmelCase							:	Union[str, List[str]] = None     , **UpperCAmelCase							:	Dict     , )	->       Any:
											if "text_queries" in kwargs:
																	lowerCAmelCase   :Dict						=			kwargs.pop('text_queries'   )

											if isinstance(UpperCAmelCase     , (str, Image.Image)   ):
																	lowerCAmelCase   :List[str]						=			{'image': image, 'candidate_labels': candidate_labels}
											else:
																	lowerCAmelCase   :Union[str, Any]						=			image
											lowerCAmelCase   :Optional[int]						=			super().__call__(UpperCAmelCase     , **UpperCAmelCase   )
											return results
					def        UpperCAmelCase__ (							self							:	List[Any]     , **UpperCAmelCase							:	Tuple   )	->       Optional[int]:
											lowerCAmelCase   :Any						=			{}
											if "threshold" in kwargs:
																	lowerCAmelCase   :int						=			kwargs['threshold']
											if "top_k" in kwargs:
																	lowerCAmelCase   :str						=			kwargs['top_k']
											return {}, {}, postprocess_params
					def        UpperCAmelCase__ (							self							:	Any     , UpperCAmelCase							:	Union[str, Any]   )	->       str:
											lowerCAmelCase   :Dict						=			load_image(inputs['image']   )
											lowerCAmelCase   :Dict						=			inputs['candidate_labels']
											if isinstance(UpperCAmelCase     , UpperCAmelCase   ):
																	lowerCAmelCase   :str						=			candidate_labels.split(','   )

											lowerCAmelCase   :str						=			torch.tensor([[image.height, image.width]]     , dtype=torch.intaa   )
											for i, candidate_label in enumerate(UpperCAmelCase   ):
																	lowerCAmelCase   :Any						=			self.tokenizer(UpperCAmelCase     , return_tensors=self.framework   )
																	lowerCAmelCase   :Optional[Any]						=			self.image_processor(UpperCAmelCase     , return_tensors=self.framework   )
																	yield {
																	    "is_last": i == len(UpperCAmelCase   ) - 1,
																	    "target_size": target_size,
																	    "candidate_label": candidate_label,
																	    **text_inputs,
																	    **image_features,
																	}
					def        UpperCAmelCase__ (							self							:	Union[str, Any]     , UpperCAmelCase							:	str   )	->       List[Any]:
											lowerCAmelCase   :Union[str, Any]						=			model_inputs.pop('target_size'   )
											lowerCAmelCase   :List[str]						=			model_inputs.pop('candidate_label'   )
											lowerCAmelCase   :Union[str, Any]						=			model_inputs.pop('is_last'   )

											lowerCAmelCase   :int						=			self.model(**UpperCAmelCase   )

											lowerCAmelCase   :Any						=			{'target_size': target_size, 'candidate_label': candidate_label, 'is_last': is_last, **outputs}
											return model_outputs
					def        UpperCAmelCase__ (							self							:	Union[str, Any]     , UpperCAmelCase							:	Optional[int]     , UpperCAmelCase							:	int=0.1     , UpperCAmelCase							:	Optional[Any]=None   )	->       str:
											lowerCAmelCase   :int						=			[]
											for model_output in model_outputs:
																	lowerCAmelCase   :int						=			model_output['candidate_label']
																	lowerCAmelCase   :List[Any]						=			BaseModelOutput(UpperCAmelCase   )
																	lowerCAmelCase   :Dict						=			self.image_processor.post_process_object_detection(
																	    outputs=UpperCAmelCase     , threshold=UpperCAmelCase     , target_sizes=model_output['target_size']   )[0]

																	for index in outputs["scores"].nonzero():
																							lowerCAmelCase   :Tuple						=			outputs['scores'][index].item()
																							lowerCAmelCase   :int						=			self._get_bounding_box(outputs['boxes'][index][0]   )

																							lowerCAmelCase   :Optional[int]						=			{'score': score, 'label': label, 'box': box}
																							results.append(UpperCAmelCase   )

											lowerCAmelCase   :Tuple						=			sorted(UpperCAmelCase     , key=lambda UpperCAmelCase   : x["score"]     , reverse=UpperCAmelCase   )
											if top_k:
																	lowerCAmelCase   :Dict						=			results[:top_k]

											return results
					def        UpperCAmelCase__ (							self							:	Optional[Any]     , UpperCAmelCase							:	"torch.Tensor"   )	->       Dict[str, int]:
											if self.framework != "pt":
																	raise ValueError('The ZeroShotObjectDetectionPipeline is only available in PyTorch.'   )
											lowerCAmelCase	, lowerCAmelCase	, lowerCAmelCase	, lowerCAmelCase   :Optional[int]						=			box.int().tolist()
											lowerCAmelCase   :Union[str, Any]						=			{
											    'xmin': xmin,
											    'ymin': ymin,
											    'xmax': xmax,
											    'ymax': ymax,
											}
											return bbox