



import unittest

import requests

from transformers import MODEL_FOR_VISION_2_SEQ_MAPPING, TF_MODEL_FOR_VISION_2_SEQ_MAPPING, is_vision_available
from transformers.pipelines import pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    is_torch_available,
    require_tf,
    require_torch,
    require_vision,
    slow,
)

from .test_pipelines_common import ANY


if is_torch_available():
	from transformers.pytorch_utils import is_torch_greater_or_equal_than_1_11
else:
	A__     				=	False


if is_vision_available():
	from PIL import Image
else:





	class  __lowerCAmelCase			:




					@staticmethod
					def 							snake_case       (							*_snake_case						,    **_snake_case		):



										"""simple docstring"""





										pass





@is_pipeline_test
@require_vision
class  __lowerCAmelCase			(					unittest.TestCase	):
				__lowerCamelCase			=    MODEL_FOR_VISION_2_SEQ_MAPPING
				__lowerCamelCase			=    TF_MODEL_FOR_VISION_2_SEQ_MAPPING




				def 							snake_case       (							self						,    _snake_case						,    _snake_case						,    _snake_case		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model=_snake_case						,    tokenizer=_snake_case						,    image_processor=_snake_case		)
									_lowerCAmelCase      =   [
									    Image.open("""./tests/fixtures/tests_samples/COCO/000000039769.png"""		),
									    """./tests/fixtures/tests_samples/COCO/000000039769.png""",
									]
									return pipe, examples




				def 							snake_case       (							self						,    _snake_case						,    _snake_case		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipe(_snake_case		)
									self.assertEqual(
									    _snake_case						,    [
									        [{"""generated_text""": ANY(_snake_case		)}],
									        [{"""generated_text""": ANY(_snake_case		)}],
									    ]						,    )




				@require_tf
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""hf-internal-testing/tiny-random-vit-gpt2"""						,    framework="""tf"""		)
									_lowerCAmelCase      =   """./tests/fixtures/tests_samples/COCO/000000039769.png"""

									_lowerCAmelCase      =   pipe(_snake_case		)
									self.assertEqual(
									    _snake_case						,    [
									        {
									            """generated_text""": """growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"""
									        },
									    ]						,    )

									_lowerCAmelCase      =   pipe([image, image]		)
									self.assertEqual(
									    _snake_case						,    [
									        [
									            {
									                """generated_text""": """growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"""
									            }
									        ],
									        [
									            {
									                """generated_text""": """growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"""
									            }
									        ],
									    ]						,    )

									_lowerCAmelCase      =   pipe(_snake_case						,    max_new_tokens=1		)
									self.assertEqual(
									    _snake_case						,    [{"""generated_text""": """growth"""}]						,    )




				@require_torch
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""hf-internal-testing/tiny-random-vit-gpt2"""		)
									_lowerCAmelCase      =   """./tests/fixtures/tests_samples/COCO/000000039769.png"""

									_lowerCAmelCase      =   pipe(_snake_case		)
									self.assertEqual(
									    _snake_case						,    [
									        {
									            """generated_text""": """growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"""
									        },
									    ]						,    )

									_lowerCAmelCase      =   pipe([image, image]		)
									self.assertEqual(
									    _snake_case						,    [
									        [
									            {
									                """generated_text""": """growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"""
									            }
									        ],
									        [
									            {
									                """generated_text""": """growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"""
									            }
									        ],
									    ]						,    )




				@require_torch
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""hf-internal-testing/tiny-random-BlipForConditionalGeneration"""		)
									_lowerCAmelCase      =   """./tests/fixtures/tests_samples/COCO/000000039769.png"""
									_lowerCAmelCase      =   """a photo of"""

									_lowerCAmelCase      =   pipe(_snake_case						,    prompt=_snake_case		)
									self.assertTrue(outputs[0]["""generated_text"""].startswith(_snake_case		)		)




				@slow
				@require_torch
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""ydshieh/vit-gpt2-coco-en"""		)
									_lowerCAmelCase      =   """./tests/fixtures/tests_samples/COCO/000000039769.png"""

									_lowerCAmelCase      =   pipe(_snake_case		)
									self.assertEqual(_snake_case						,    [{"""generated_text""": """a cat laying on a blanket next to a cat laying on a bed """}]		)

									_lowerCAmelCase      =   pipe([image, image]		)
									self.assertEqual(
									    _snake_case						,    [
									        [{"""generated_text""": """a cat laying on a blanket next to a cat laying on a bed """}],
									        [{"""generated_text""": """a cat laying on a blanket next to a cat laying on a bed """}],
									    ]						,    )




				@slow
				@require_torch
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""Salesforce/blip-image-captioning-base"""		)
									_lowerCAmelCase      =   """https://huggingface.co/datasets/sayakpaul/sample-datasets/resolve/main/pokemon.png"""
									_lowerCAmelCase      =   Image.open(requests.get(_snake_case						,    stream=_snake_case		).raw		)

									_lowerCAmelCase      =   pipe(_snake_case		)
									self.assertEqual(_snake_case						,    [{"""generated_text""": """a pink pokemon pokemon with a blue shirt and a blue shirt"""}]		)




				@slow
				@require_torch
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""microsoft/git-base-coco"""		)
									_lowerCAmelCase      =   """https://huggingface.co/datasets/sayakpaul/sample-datasets/resolve/main/pokemon.png"""
									_lowerCAmelCase      =   Image.open(requests.get(_snake_case						,    stream=_snake_case		).raw		)

									_lowerCAmelCase      =   pipe(_snake_case		)
									self.assertEqual(_snake_case						,    [{"""generated_text""": """a cartoon of a purple character."""}]		)




				@slow
				@require_torch
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""Salesforce/blip-image-captioning-base"""		)
									_lowerCAmelCase      =   """https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"""
									_lowerCAmelCase      =   Image.open(requests.get(_snake_case						,    stream=_snake_case		).raw		)

									_lowerCAmelCase      =   """a photography of"""

									_lowerCAmelCase      =   pipe(_snake_case						,    prompt=_snake_case		)
									self.assertEqual(_snake_case						,    [{"""generated_text""": """a photography of a volcano"""}]		)

									with self.assertRaises(_snake_case		):
														_lowerCAmelCase      =   pipe([image, image]						,    prompt=[prompt, prompt]		)




				@slow
				@require_torch
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""microsoft/git-base-coco"""		)
									_lowerCAmelCase      =   """https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"""
									_lowerCAmelCase      =   Image.open(requests.get(_snake_case						,    stream=_snake_case		).raw		)

									_lowerCAmelCase      =   """a photo of a"""

									_lowerCAmelCase      =   pipe(_snake_case						,    prompt=_snake_case		)
									self.assertEqual(_snake_case						,    [{"""generated_text""": """a photo of a tent with a tent and a tent in the background."""}]		)

									with self.assertRaises(_snake_case		):
														_lowerCAmelCase      =   pipe([image, image]						,    prompt=[prompt, prompt]		)




				@unittest.skipIf(
				    not is_torch_greater_or_equal_than_1_11						,    reason="""`Pix2StructImageProcessor` requires `torch>=1.11.0`."""		)
				@slow
				@require_torch
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""google/pix2struct-ai2d-base"""		)
									_lowerCAmelCase      =   """https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"""
									_lowerCAmelCase      =   Image.open(requests.get(_snake_case						,    stream=_snake_case		).raw		)

									_lowerCAmelCase      =   """What does the label 15 represent? (1) lava (2) core (3) tunnel (4) ash cloud"""

									_lowerCAmelCase      =   pipe(_snake_case						,    prompt=_snake_case		)
									self.assertEqual(_snake_case						,    [{"""generated_text""": """ash cloud"""}]		)

									with self.assertRaises(_snake_case		):
														_lowerCAmelCase      =   pipe([image, image]						,    prompt=[prompt, prompt]		)




				@slow
				@require_tf
				def 							snake_case       (							self		):



									"""simple docstring"""





									_lowerCAmelCase      =   pipeline("""image-to-text"""						,    model="""ydshieh/vit-gpt2-coco-en"""						,    framework="""tf"""		)
									_lowerCAmelCase      =   """./tests/fixtures/tests_samples/COCO/000000039769.png"""

									_lowerCAmelCase      =   pipe(_snake_case		)
									self.assertEqual(_snake_case						,    [{"""generated_text""": """a cat laying on a blanket next to a cat laying on a bed """}]		)

									_lowerCAmelCase      =   pipe([image, image]		)
									self.assertEqual(
									    _snake_case						,    [
									        [{"""generated_text""": """a cat laying on a blanket next to a cat laying on a bed """}],
									        [{"""generated_text""": """a cat laying on a blanket next to a cat laying on a bed """}],
									    ]						,    )








from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


A__     				=	logging.get_logger(__name__)





class  __lowerCAmelCase			(					lowerCamelCase__	):
				__lowerCamelCase			=    ['''input_values''', '''attention_mask''']




				def __init__(							self						,    _snake_case=1						,    _snake_case=16000						,    _snake_case=128						,    _snake_case=1024						,    _snake_case=0.0						,    _snake_case=True						,    _snake_case=-4.267_7393						,    _snake_case=4.568_9974						,    _snake_case=False						,    **_snake_case						,    ):



									"""simple docstring"""





									super().__init__(feature_size=_snake_case						,    sampling_rate=_snake_case						,    padding_value=_snake_case						,    **_snake_case		)
									_lowerCAmelCase      =   num_mel_bins
									_lowerCAmelCase      =   max_length
									_lowerCAmelCase      =   do_normalize
									_lowerCAmelCase      =   mean
									_lowerCAmelCase      =   std
									_lowerCAmelCase      =   return_attention_mask




				def 							snake_case       (							self						,    _snake_case						,    _snake_case						,    ):



									"""simple docstring"""





									_lowerCAmelCase      =   torch.from_numpy(_snake_case		).unsqueeze(0		)
									_lowerCAmelCase      =   ta_kaldi.fbank(
									    _snake_case						,    htk_compat=_snake_case						,    sample_frequency=self.sampling_rate						,    use_energy=_snake_case						,    window_type="""hanning"""						,    num_mel_bins=self.num_mel_bins						,    dither=0.0						,    frame_shift=10						,    )

									_lowerCAmelCase      =   fbank.shape[0]
									_lowerCAmelCase      =   max_length - n_frames

									# pad or truncate, depending on difference
									if difference > 0:
														_lowerCAmelCase      =   torch.nn.ZeroPadad((0, 0, 0, difference)		)
														_lowerCAmelCase      =   pad_module(_snake_case		)
									elif difference < 0:
														_lowerCAmelCase      =   fbank[0:max_length, :]

									_lowerCAmelCase      =   fbank.numpy()

									return fbank




				def 							snake_case       (							self						,    _snake_case		):



									"""simple docstring"""





									return (input_values - (self.mean)) / (self.std * 2)




				def __call__(							self						,    _snake_case						,    _snake_case = None						,    _snake_case = None						,    **_snake_case						,    ):



									"""simple docstring"""





									if sampling_rate is not None:
														if sampling_rate != self.sampling_rate:
																			raise ValueError(
																			    F'The model corresponding to this feature extractor: {self} was trained using a sampling rate of'
																			    F' {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with'
																			    F' {self.sampling_rate} and not {sampling_rate}.'		)
									else:
														logger.warning(
														    """It is strongly recommended to pass the `sampling_rate` argument to this function. """
														    """Failing to do so can result in silent errors that might be hard to debug."""		)

									_lowerCAmelCase      =   isinstance(_snake_case						,    np.ndarray		) and len(raw_speech.shape		) > 1
									if is_batched_numpy and len(raw_speech.shape		) > 2:
														raise ValueError(F'Only mono-channel audio is supported for input to {self}'		)
									_lowerCAmelCase      =   is_batched_numpy or (
									    isinstance(_snake_case						,    (list, tuple)		) and (isinstance(raw_speech[0]						,    (np.ndarray, tuple, list)		))
									)

									if is_batched:
														_lowerCAmelCase      =   [np.asarray(_snake_case						,    dtype=np.floataa		) for speech in raw_speech]
									elif not is_batched and not isinstance(_snake_case						,    np.ndarray		):
														_lowerCAmelCase      =   np.asarray(_snake_case						,    dtype=np.floataa		)
									elif isinstance(_snake_case						,    np.ndarray		) and raw_speech.dtype is np.dtype(np.floataa		):
														_lowerCAmelCase      =   raw_speech.astype(np.floataa		)

									# always return batch
									if not is_batched:
														_lowerCAmelCase      =   [raw_speech]

									# extract fbank features and pad/truncate to max_length
									_lowerCAmelCase      =   [self._extract_fbank_features(_snake_case						,    max_length=self.max_length		) for waveform in raw_speech]

									# convert into BatchFeature
									_lowerCAmelCase      =   BatchFeature({"""input_values""": features}		)

									# make sure list is in array format
									_lowerCAmelCase      =   padded_inputs.get("""input_values"""		)
									if isinstance(input_values[0]						,    _snake_case		):
														_lowerCAmelCase      =   [np.asarray(_snake_case						,    dtype=np.floataa		) for feature in input_values]

									# normalization
									if self.do_normalize:
														_lowerCAmelCase      =   [self.normalize(_snake_case		) for feature in input_values]

									if return_tensors is not None:
														_lowerCAmelCase      =   padded_inputs.convert_to_tensors(_snake_case		)

									return padded_inputs



