

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 							A_			:


       @staticmethod
       def   UpperCAmelCase  (							*UpperCAmelCase      :		Tuple      ,  **UpperCAmelCase      :		Dict )							->     Optional[Any]:
        pass






@is_pipeline_test
@require_vision
class 							A_			(		unittest.TestCase ):
     _lowercase      :					Any       =     MODEL_FOR_VISION_2_SEQ_MAPPING
     _lowercase      :					Dict       =     TF_MODEL_FOR_VISION_2_SEQ_MAPPING


     def   UpperCAmelCase  (							self      :		int      ,  UpperCAmelCase      :		List[Any]      ,  UpperCAmelCase      :		Union[str, Any]      ,  UpperCAmelCase      :		str )							->     List[Any]:
      __lowerCAmelCase:       str            =				pipeline('image-to-text'      ,  model=UpperCAmelCase      ,  tokenizer=UpperCAmelCase      ,  image_processor=UpperCAmelCase )
      __lowerCAmelCase:       Any            =				[
          Image.open('./tests/fixtures/tests_samples/COCO/000000039769.png' ),
          './tests/fixtures/tests_samples/COCO/000000039769.png',
      ]
      return pipe, examples


     def   UpperCAmelCase  (							self      :		str      ,  UpperCAmelCase      :		List[Any]      ,  UpperCAmelCase      :		Tuple )							->     List[Any]:
      __lowerCAmelCase:       str            =				pipe(UpperCAmelCase )
      self.assertEqual(
          UpperCAmelCase      ,  [
              [{'generated_text': ANY(UpperCAmelCase )}],
              [{'generated_text': ANY(UpperCAmelCase )}],
          ]      ,  )


     @require_tf
     def   UpperCAmelCase  (							self      :		Optional[Any] )							->     List[str]:
      __lowerCAmelCase:       Tuple            =				pipeline('image-to-text'      ,  model='hf-internal-testing/tiny-random-vit-gpt2'      ,  framework='tf' )
      __lowerCAmelCase:       List[Any]            =				'./tests/fixtures/tests_samples/COCO/000000039769.png'

      __lowerCAmelCase:       Optional[int]            =				pipe(UpperCAmelCase )
      self.assertEqual(
          UpperCAmelCase      ,  [
              {
                  'generated_text': 'growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'
              },
          ]      ,  )

      __lowerCAmelCase:       Optional[int]            =				pipe([image, image] )
      self.assertEqual(
          UpperCAmelCase      ,  [
              [
                  {
                      'generated_text': 'growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'
                  }
              ],
              [
                  {
                      'generated_text': 'growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'
                  }
              ],
          ]      ,  )

      __lowerCAmelCase:       int            =				pipe(UpperCAmelCase      ,  max_new_tokens=1 )
      self.assertEqual(
          UpperCAmelCase      ,  [{'generated_text': 'growth'}]      ,  )


     @require_torch
     def   UpperCAmelCase  (							self      :		List[Any] )							->     str:
      __lowerCAmelCase:       int            =				pipeline('image-to-text'      ,  model='hf-internal-testing/tiny-random-vit-gpt2' )
      __lowerCAmelCase:       List[Any]            =				'./tests/fixtures/tests_samples/COCO/000000039769.png'

      __lowerCAmelCase:       Any            =				pipe(UpperCAmelCase )
      self.assertEqual(
          UpperCAmelCase      ,  [
              {
                  'generated_text': 'growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'
              },
          ]      ,  )

      __lowerCAmelCase:       Dict            =				pipe([image, image] )
      self.assertEqual(
          UpperCAmelCase      ,  [
              [
                  {
                      'generated_text': 'growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'
                  }
              ],
              [
                  {
                      'generated_text': 'growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'
                  }
              ],
          ]      ,  )


     @require_torch
     def   UpperCAmelCase  (							self      :		Union[str, Any] )							->     Tuple:
      __lowerCAmelCase:       int            =				pipeline('image-to-text'      ,  model='hf-internal-testing/tiny-random-BlipForConditionalGeneration' )
      __lowerCAmelCase:       Optional[Any]            =				'./tests/fixtures/tests_samples/COCO/000000039769.png'
      __lowerCAmelCase:       List[Any]            =				'a photo of'

      __lowerCAmelCase:       Dict            =				pipe(UpperCAmelCase      ,  prompt=UpperCAmelCase )
      self.assertTrue(outputs[0]['generated_text'].startswith(UpperCAmelCase ) )


     @slow
     @require_torch
     def   UpperCAmelCase  (							self      :		Tuple )							->     List[Any]:
      __lowerCAmelCase:       int            =				pipeline('image-to-text'      ,  model='ydshieh/vit-gpt2-coco-en' )
      __lowerCAmelCase:       Optional[int]            =				'./tests/fixtures/tests_samples/COCO/000000039769.png'

      __lowerCAmelCase:       str            =				pipe(UpperCAmelCase )
      self.assertEqual(UpperCAmelCase      ,  [{'generated_text': 'a cat laying on a blanket next to a cat laying on a bed '}] )

      __lowerCAmelCase:       Dict            =				pipe([image, image] )
      self.assertEqual(
          UpperCAmelCase      ,  [
              [{'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   UpperCAmelCase  (							self      :		Any )							->     List[str]:
      __lowerCAmelCase:       Any            =				pipeline('image-to-text'      ,  model='Salesforce/blip-image-captioning-base' )
      __lowerCAmelCase:       Optional[Any]            =				'https://huggingface.co/datasets/sayakpaul/sample-datasets/resolve/main/pokemon.png'
      __lowerCAmelCase:       str            =				Image.open(requests.get(UpperCAmelCase      ,  stream=UpperCAmelCase ).raw )

      __lowerCAmelCase:       Tuple            =				pipe(UpperCAmelCase )
      self.assertEqual(UpperCAmelCase      ,  [{'generated_text': 'a pink pokemon pokemon with a blue shirt and a blue shirt'}] )


     @slow
     @require_torch
     def   UpperCAmelCase  (							self      :		List[str] )							->     Dict:
      __lowerCAmelCase:       List[Any]            =				pipeline('image-to-text'      ,  model='microsoft/git-base-coco' )
      __lowerCAmelCase:       List[Any]            =				'https://huggingface.co/datasets/sayakpaul/sample-datasets/resolve/main/pokemon.png'
      __lowerCAmelCase:       int            =				Image.open(requests.get(UpperCAmelCase      ,  stream=UpperCAmelCase ).raw )

      __lowerCAmelCase:       Dict            =				pipe(UpperCAmelCase )
      self.assertEqual(UpperCAmelCase      ,  [{'generated_text': 'a cartoon of a purple character.'}] )


     @slow
     @require_torch
     def   UpperCAmelCase  (							self      :		Union[str, Any] )							->     int:
      __lowerCAmelCase:       Optional[Any]            =				pipeline('image-to-text'      ,  model='Salesforce/blip-image-captioning-base' )
      __lowerCAmelCase:       Optional[Any]            =				'https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg'
      __lowerCAmelCase:       Any            =				Image.open(requests.get(UpperCAmelCase      ,  stream=UpperCAmelCase ).raw )

      __lowerCAmelCase:       Optional[int]            =				'a photography of'

      __lowerCAmelCase:       int            =				pipe(UpperCAmelCase      ,  prompt=UpperCAmelCase )
      self.assertEqual(UpperCAmelCase      ,  [{'generated_text': 'a photography of a volcano'}] )

      with self.assertRaises(UpperCAmelCase ):
       __lowerCAmelCase:       List[str]            =				pipe([image, image]      ,  prompt=[prompt, prompt] )


     @slow
     @require_torch
     def   UpperCAmelCase  (							self      :		Union[str, Any] )							->     Optional[int]:
      __lowerCAmelCase:       str            =				pipeline('image-to-text'      ,  model='microsoft/git-base-coco' )
      __lowerCAmelCase:       Dict            =				'https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg'
      __lowerCAmelCase:       Optional[int]            =				Image.open(requests.get(UpperCAmelCase      ,  stream=UpperCAmelCase ).raw )

      __lowerCAmelCase:       Tuple            =				'a photo of a'

      __lowerCAmelCase:       Dict            =				pipe(UpperCAmelCase      ,  prompt=UpperCAmelCase )
      self.assertEqual(UpperCAmelCase      ,  [{'generated_text': 'a photo of a tent with a tent and a tent in the background.'}] )

      with self.assertRaises(UpperCAmelCase ):
       __lowerCAmelCase:       Union[str, Any]            =				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   UpperCAmelCase  (							self      :		List[Any] )							->     List[str]:
      __lowerCAmelCase:       int            =				pipeline('image-to-text'      ,  model='google/pix2struct-ai2d-base' )
      __lowerCAmelCase:       Optional[Any]            =				'https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg'
      __lowerCAmelCase:       List[str]            =				Image.open(requests.get(UpperCAmelCase      ,  stream=UpperCAmelCase ).raw )

      __lowerCAmelCase:       List[str]            =				'What does the label 15 represent? (1) lava (2) core (3) tunnel (4) ash cloud'

      __lowerCAmelCase:       Union[str, Any]            =				pipe(UpperCAmelCase      ,  prompt=UpperCAmelCase )
      self.assertEqual(UpperCAmelCase      ,  [{'generated_text': 'ash cloud'}] )

      with self.assertRaises(UpperCAmelCase ):
       __lowerCAmelCase:       Tuple            =				pipe([image, image]      ,  prompt=[prompt, prompt] )


     @slow
     @require_tf
     def   UpperCAmelCase  (							self      :		int )							->     Tuple:
      __lowerCAmelCase:       Union[str, Any]            =				pipeline('image-to-text'      ,  model='ydshieh/vit-gpt2-coco-en'      ,  framework='tf' )
      __lowerCAmelCase:       int            =				'./tests/fixtures/tests_samples/COCO/000000039769.png'

      __lowerCAmelCase:       int            =				pipe(UpperCAmelCase )
      self.assertEqual(UpperCAmelCase      ,  [{'generated_text': 'a cat laying on a blanket next to a cat laying on a bed '}] )

      __lowerCAmelCase:       Dict            =				pipe([image, image] )
      self.assertEqual(
          UpperCAmelCase      ,  [
              [{'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 __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass
from typing import Any






class 							A_			:
     _lowercase      :					Any       =     ('_front', '_back', '_len')






     @dataclass
     class 							A_			:
          _lowercase      :					Any       =     None
          _lowercase      :					Deque._Node | None       =     None
          _lowercase      :					Deque._Node | None       =     None







     class 							A_			:
          _lowercase      :					Optional[Any]       =     ('_cur',)


          def __init__(							self      :		Any      ,  UpperCAmelCase      :		Deque._Node | None )							->     None:
           __lowerCAmelCase:       int            =				cur


          def __iter__(							self      :		int )							->     Deque._Iterator:
           return self


          def   UpperCAmelCase  (							self      :		Optional[Any] )							->     Any:
           if self._cur is None:
            # finished iterating
            raise StopIteration
           __lowerCAmelCase:       Optional[Any]            =				self._cur.val
           __lowerCAmelCase:       List[Any]            =				self._cur.next_node

           return val







     def __init__(							self      :		Any      ,  UpperCAmelCase      :		Iterable[Any] | None = None )							->     None:
      __lowerCAmelCase:       Any            =				None
      __lowerCAmelCase:       Any            =				None
      __lowerCAmelCase:       int            =				0

      if iterable is not None:
       # append every value to the deque
       for val in iterable:
        self.append(UpperCAmelCase )


     def   UpperCAmelCase  (							self      :		List[str]      ,  UpperCAmelCase      :		Any )							->     None:
      __lowerCAmelCase:       Union[str, Any]            =				self._Node(UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase )
      if self.is_empty():
       # front = back
       __lowerCAmelCase:       int            =				node
       __lowerCAmelCase:       List[Any]            =				1
      else:
       # connect nodes
       __lowerCAmelCase:       Optional[Any]            =				node
       __lowerCAmelCase:       str            =				self._back
       __lowerCAmelCase:       Optional[int]            =				node  # assign new back to the new node

       self._len += 1

       # make sure there were no errors
       assert not self.is_empty(), "Error on appending value."


     def   UpperCAmelCase  (							self      :		Optional[int]      ,  UpperCAmelCase      :		Any )							->     None:
      __lowerCAmelCase:       int            =				self._Node(UpperCAmelCase      ,  UpperCAmelCase      ,  UpperCAmelCase )
      if self.is_empty():
       # front = back
       __lowerCAmelCase:       Union[str, Any]            =				node
       __lowerCAmelCase:       str            =				1
      else:
       # connect nodes
       __lowerCAmelCase:       Dict            =				self._front
       __lowerCAmelCase:       Optional[int]            =				node
       __lowerCAmelCase:       Any            =				node  # assign new front to the new node

       self._len += 1

       # make sure there were no errors
       assert not self.is_empty(), "Error on appending value."


     def   UpperCAmelCase  (							self      :		Union[str, Any]      ,  UpperCAmelCase      :		Iterable[Any] )							->     None:
      for val in iterable:
       self.append(UpperCAmelCase )


     def   UpperCAmelCase  (							self      :		List[str]      ,  UpperCAmelCase      :		Iterable[Any] )							->     None:
      for val in iterable:
       self.appendleft(UpperCAmelCase )


     def   UpperCAmelCase  (							self      :		Dict )							->     Any:
      assert not self.is_empty(), "Deque is empty."

      __lowerCAmelCase:       Optional[int]            =				self._back
      __lowerCAmelCase:       Optional[Any]            =				self._back.prev_node  # set new back
      # drop the last node - python will deallocate memory automatically
      __lowerCAmelCase:       Optional[int]            =				None

      self._len -= 1

      return topop.val


     def   UpperCAmelCase  (							self      :		Dict )							->     Any:
      assert not self.is_empty(), "Deque is empty."

      __lowerCAmelCase:       Optional[Any]            =				self._front
      __lowerCAmelCase:       Optional[Any]            =				self._front.next_node  # set new front and drop the first node
      __lowerCAmelCase:       int            =				None

      self._len -= 1

      return topop.val


     def   UpperCAmelCase  (							self      :		Dict )							->     bool:
      return self._front is None


     def __len__(							self      :		List[Any] )							->     int:
      return self._len


     def __eq__(							self      :		Union[str, Any]      ,  UpperCAmelCase      :		object )							->     bool:

      if not isinstance(UpperCAmelCase      ,  UpperCAmelCase ):
       return NotImplemented

      __lowerCAmelCase:       List[Any]            =				self._front
      __lowerCAmelCase:       Dict            =				other._front

      # if the length of the dequeues are not the same, they are not equal
      if len(self ) != len(UpperCAmelCase ):
       return False

      while me is not None and oth is not None:
       # compare every value
       if me.val != oth.val:
        return False
       __lowerCAmelCase:       Optional[Any]            =				me.next_node
       __lowerCAmelCase:       Optional[Any]            =				oth.next_node

      return True


     def __iter__(							self      :		Tuple )							->     Deque._Iterator:
      return Deque._Iterator(self._front )


     def __repr__(							self      :		Optional[Any] )							->     str:
      __lowerCAmelCase:       Union[str, Any]            =				[]
      __lowerCAmelCase:       Optional[int]            =				self._front
      while aux is not None:
       # append the values in a list to display
       values_list.append(aux.val )
       __lowerCAmelCase:       Any            =				aux.next_node

      return F'''[{', '.join(repr(UpperCAmelCase ) for val in values_list )}]'''


if __name__ == "__main__":
  import doctest

  doctest.testmod()





