



import itertools
import random
import unittest

import numpy as np

from transformers import is_speech_available
from transformers.testing_utils import require_torch, require_torchaudio

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
     from transformers import SpeechaTextFeatureExtractor

a_      :      List[str]    =     random.Random()






def     __a					(		__UpperCAmelCase							,  __UpperCAmelCase=1.0							,  __UpperCAmelCase=None							,  __UpperCAmelCase=None       ):
 if rng is None:
  a__												= global_rng

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

 return values
@require_torch
@require_torchaudio
class    __UpperCamelCase			(       unittest.TestCase				):

 """simple docstring"""



 def __init__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=7		,				SCREAMING_SNAKE_CASE=4_0_0		,				SCREAMING_SNAKE_CASE=2_0_0_0		,				SCREAMING_SNAKE_CASE=2_4		,				SCREAMING_SNAKE_CASE=2_4		,				SCREAMING_SNAKE_CASE=0.0		,				SCREAMING_SNAKE_CASE=1_6_0_0_0		,				SCREAMING_SNAKE_CASE=True		,				SCREAMING_SNAKE_CASE=True		,				) ->			int:
  a__												= parent
  a__												= batch_size
  a__												= min_seq_length
  a__												= max_seq_length
  a__												= (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
  a__												= feature_size
  a__												= num_mel_bins
  a__												= padding_value
  a__												= sampling_rate
  a__												= return_attention_mask
  a__												= do_normalize

 def 						_UpperCAmelCase       (	self      ) ->			Tuple:
  return {
      "feature_size": self.feature_size,
      "num_mel_bins": self.num_mel_bins,
      "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		,				SCREAMING_SNAKE_CASE=False		,				SCREAMING_SNAKE_CASE=False      ) ->			Tuple:
  def _flatten(SCREAMING_SNAKE_CASE      ):
   return list(itertools.chain(*SCREAMING_SNAKE_CASE      )      )

  if equal_length:
   a__												= [floats_list((self.max_seq_length, self.feature_size)      ) for _ in range(self.batch_size      )]
  else:
   # make sure that inputs increase in size
   a__												= [
       floats_list((x, self.feature_size)      )
       for x in range(self.min_seq_length		,				self.max_seq_length		,				self.seq_length_diff      )
   ]
  if numpify:
   a__												= [np.asarray(SCREAMING_SNAKE_CASE      ) for x in speech_inputs]
  return speech_inputs





@require_torch
@require_torchaudio
class    __UpperCamelCase			(       _lowercase	,	unittest.TestCase				):

 """simple docstring"""



 _lowercase   :    List[Any]      						=					SpeechaTextFeatureExtractor if is_speech_available() else None

 def 						_UpperCAmelCase       (	self      ) ->			int:
  a__												= SpeechaTextFeatureExtractionTester(self      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			str:
  self.assertTrue(np.all(np.mean(SCREAMING_SNAKE_CASE		,				axis=0      ) < 1e-3      )      )
  self.assertTrue(np.all(np.abs(np.var(SCREAMING_SNAKE_CASE		,				axis=0      ) - 1      ) < 1e-3      )      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  # Tests that all call wrap to encode_plus and batch_encode_plus
  a__												= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
  # create three inputs of length 800, 1000, and 1200
  a__												= [floats_list((1, x)      )[0] for x in range(8_0_0		,				1_4_0_0		,				2_0_0      )]
  a__												= [np.asarray(SCREAMING_SNAKE_CASE      ) for speech_input in speech_inputs]

  # Test feature size
  a__												= feature_extractor(SCREAMING_SNAKE_CASE		,				padding=SCREAMING_SNAKE_CASE		,				return_tensors='''np'''      ).input_features
  self.assertTrue(input_features.ndim == 3      )
  self.assertTrue(input_features.shape[-1] == feature_extractor.feature_size      )

  # Test not batched input
  a__												= feature_extractor(speech_inputs[0]		,				return_tensors='''np'''      ).input_features
  a__												= feature_extractor(np_speech_inputs[0]		,				return_tensors='''np'''      ).input_features
  self.assertTrue(np.allclose(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				atol=1e-3      )      )

  # Test batched
  a__												= feature_extractor(SCREAMING_SNAKE_CASE		,				return_tensors='''np'''      ).input_features
  a__												= feature_extractor(SCREAMING_SNAKE_CASE		,				return_tensors='''np'''      ).input_features
  for enc_seq_a, enc_seq_a in zip(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   self.assertTrue(np.allclose(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				atol=1e-3      )      )

  # Test 2-D numpy arrays are batched.
  a__												= [floats_list((1, x)      )[0] for x in (8_0_0, 8_0_0, 8_0_0)]
  a__												= np.asarray(SCREAMING_SNAKE_CASE      )
  a__												= feature_extractor(SCREAMING_SNAKE_CASE		,				return_tensors='''np'''      ).input_features
  a__												= feature_extractor(SCREAMING_SNAKE_CASE		,				return_tensors='''np'''      ).input_features
  for enc_seq_a, enc_seq_a in zip(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   self.assertTrue(np.allclose(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE		,				atol=1e-3      )      )

 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  a__												= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
  a__												= [floats_list((1, x)      )[0] for x in range(8_0_0		,				1_4_0_0		,				2_0_0      )]

  a__												= ['''longest''', '''max_length''', '''do_not_pad''']
  a__												= [None, 1_6, None]
  for max_length, padding in zip(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   a__												= feature_extractor(
       SCREAMING_SNAKE_CASE		,				padding=SCREAMING_SNAKE_CASE		,				max_length=SCREAMING_SNAKE_CASE		,				return_attention_mask=SCREAMING_SNAKE_CASE      )
   a__												= inputs.input_features
   a__												= inputs.attention_mask
   a__												= [np.sum(SCREAMING_SNAKE_CASE      ) for x in attention_mask]

   self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]      )
   self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]      )
   self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]      )

 def 						_UpperCAmelCase       (	self      ) ->			Optional[Any]:
  a__												= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
  a__												= [floats_list((1, x)      )[0] for x in range(8_0_0		,				1_4_0_0		,				2_0_0      )]

  a__												= ['''longest''', '''max_length''', '''do_not_pad''']
  a__												= [None, 1_6, None]
  for max_length, padding in zip(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   a__												= feature_extractor(
       SCREAMING_SNAKE_CASE		,				max_length=SCREAMING_SNAKE_CASE		,				padding=SCREAMING_SNAKE_CASE		,				return_tensors='''np'''		,				return_attention_mask=SCREAMING_SNAKE_CASE      )
   a__												= inputs.input_features
   a__												= inputs.attention_mask
   a__												= [np.sum(SCREAMING_SNAKE_CASE      ) for x in attention_mask]

   self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]      )
   self.assertTrue(input_features[0][fbank_feat_lengths[0] :].sum() < 1e-6      )
   self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]      )
   self.assertTrue(input_features[0][fbank_feat_lengths[1] :].sum() < 1e-6      )
   self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]      )

 def 						_UpperCAmelCase       (	self      ) ->			List[Any]:
  a__												= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
  a__												= [floats_list((1, x)      )[0] for x in range(8_0_0		,				1_4_0_0		,				2_0_0      )]
  a__												= feature_extractor(
      SCREAMING_SNAKE_CASE		,				padding='''max_length'''		,				max_length=4		,				truncation=SCREAMING_SNAKE_CASE		,				return_tensors='''np'''		,				return_attention_mask=SCREAMING_SNAKE_CASE		,				)
  a__												= inputs.input_features
  a__												= inputs.attention_mask
  a__												= np.sum(attention_mask == 1		,				axis=1      )

  self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]      )
  self._check_zero_mean_unit_variance(input_features[1]      )
  self._check_zero_mean_unit_variance(input_features[2]      )

 def 						_UpperCAmelCase       (	self      ) ->			Tuple:
  a__												= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
  a__												= [floats_list((1, x)      )[0] for x in range(8_0_0		,				1_4_0_0		,				2_0_0      )]
  a__												= feature_extractor(
      SCREAMING_SNAKE_CASE		,				padding='''longest'''		,				max_length=4		,				truncation=SCREAMING_SNAKE_CASE		,				return_tensors='''np'''		,				return_attention_mask=SCREAMING_SNAKE_CASE		,				)
  a__												= inputs.input_features
  a__												= inputs.attention_mask
  a__												= np.sum(attention_mask == 1		,				axis=1      )

  self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]      )
  self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]      )
  self._check_zero_mean_unit_variance(input_features[2]      )

  # make sure that if max_length < longest -> then pad to max_length
  self.assertEqual(input_features.shape		,				(3, 4, 2_4)      )

  a__												= [floats_list((1, x)      )[0] for x in range(8_0_0		,				1_4_0_0		,				2_0_0      )]
  a__												= feature_extractor(
      SCREAMING_SNAKE_CASE		,				padding='''longest'''		,				max_length=1_6		,				truncation=SCREAMING_SNAKE_CASE		,				return_tensors='''np'''		,				return_attention_mask=SCREAMING_SNAKE_CASE		,				)
  a__												= inputs.input_features
  a__												= inputs.attention_mask
  a__												= np.sum(attention_mask == 1		,				axis=1      )

  self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]      )
  self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]      )
  self._check_zero_mean_unit_variance(input_features[2]      )

  # make sure that if max_length < longest -> then pad to max_length
  self.assertEqual(input_features.shape		,				(3, 6, 2_4)      )

 def 						_UpperCAmelCase       (	self      ) ->			List[str]:
  import torch

  a__												= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
  a__												= np.random.rand(1_0_0		,				3_2      ).astype(np.floataa      )
  a__												= np_speech_inputs.tolist()

  for inputs in [py_speech_inputs, np_speech_inputs]:
   a__												= feature_extractor.pad([{'''input_features''': inputs}]		,				return_tensors='''np'''      )
   self.assertTrue(np_processed.input_features.dtype == np.floataa      )
   a__												= feature_extractor.pad([{'''input_features''': inputs}]		,				return_tensors='''pt'''      )
   self.assertTrue(pt_processed.input_features.dtype == torch.floataa      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			List[Any]:
  from datasets import load_dataset

  a__												= load_dataset('''hf-internal-testing/librispeech_asr_dummy'''		,				'''clean'''		,				split='''validation'''      )
  # automatic decoding with librispeech
  a__												= ds.sort('''id'''      ).select(range(SCREAMING_SNAKE_CASE      )      )[:num_samples]['''audio''']

  return [x["array"] for x in speech_samples]



 def 						_UpperCAmelCase       (	self      ) ->			Dict:
  # fmt: off
  a__												= np.array([
      -1.57_45, -1.77_13, -1.70_20, -1.60_69, -1.22_50, -1.11_05, -0.90_72, -0.82_41,
      -1.23_10, -0.80_98, -0.33_20, -0.41_01, -0.79_85, -0.49_96, -0.82_13, -0.91_28,
      -1.04_20, -1.12_86, -1.04_40, -0.79_99, -0.84_05, -1.22_75, -1.54_43, -1.46_25,
  ]      )
  # fmt: on

  a__												= self._load_datasamples(1      )
  a__												= self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()      )
  a__												= feature_extractor(SCREAMING_SNAKE_CASE		,				return_tensors='''pt'''      ).input_features
  self.assertEquals(input_features.shape		,				(1, 5_8_4, 2_4)      )
  self.assertTrue(np.allclose(input_features[0, 0, :3_0]		,				SCREAMING_SNAKE_CASE		,				atol=1e-4      )      )










from typing import Union

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


if is_vision_available():
     from PIL import Image

     from ..image_utils import load_image

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

a_      :      str    =     logging.get_logger(__name__)
@add_end_docstrings(_lowercase				)
class    __UpperCamelCase			(       _lowercase				):

 """simple docstring"""



 def __init__(	self		,				*SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      ) ->			int:
  super().__init__(*SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )
  self.check_model_type(SCREAMING_SNAKE_CASE      )

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE=None		,				SCREAMING_SNAKE_CASE=None		,				SCREAMING_SNAKE_CASE=None		,				**SCREAMING_SNAKE_CASE      ) ->			Union[str, Any]:
  a__							,			a__												= {}, {}
  if padding is not None:
   a__												= padding
  if truncation is not None:
   a__												= truncation
  if top_k is not None:
   a__												= top_k
  return preprocess_params, {}, postprocess_params

 def __call__(	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE = None		,				**SCREAMING_SNAKE_CASE      ) ->			Optional[int]:
  if isinstance(SCREAMING_SNAKE_CASE		,				(Image.Image, str)      ) and isinstance(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      ):
   a__												= {'''image''': image, '''question''': question}
  else:
   a__												= image
  a__												= super().__call__(SCREAMING_SNAKE_CASE		,				**SCREAMING_SNAKE_CASE      )
  return results

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=False		,				SCREAMING_SNAKE_CASE=False      ) ->			Dict:
  a__												= load_image(inputs['''image''']      )
  a__												= self.tokenizer(
      inputs['''question''']		,				return_tensors=self.framework		,				padding=SCREAMING_SNAKE_CASE		,				truncation=SCREAMING_SNAKE_CASE      )
  a__												= self.image_processor(images=SCREAMING_SNAKE_CASE		,				return_tensors=self.framework      )
  model_inputs.update(SCREAMING_SNAKE_CASE      )
  return model_inputs

 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE      ) ->			Union[str, Any]:
  a__												= self.model(**SCREAMING_SNAKE_CASE      )
  return model_outputs



 def 						_UpperCAmelCase       (	self		,				SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE=5      ) ->			Optional[Any]:
  if top_k > self.model.config.num_labels:
   a__												= self.model.config.num_labels

  if self.framework == "pt":
   a__												= model_outputs.logits.sigmoid()[0]
   a__							,			a__												= probs.topk(SCREAMING_SNAKE_CASE      )
  else:
   raise ValueError(f"Unsupported framework: {self.framework}"      )

  a__												= scores.tolist()
  a__												= ids.tolist()
  return [{"score": score, "answer": self.model.config.idalabel[_id]} for score, _id in zip(SCREAMING_SNAKE_CASE		,				SCREAMING_SNAKE_CASE      )]





