




import importlib
import inspect
from functools import wraps
from typing import TYPE_CHECKING, Optional, Union

from .download.streaming_download_manager import (
    xbasename,
    xdirname,
    xet_parse,
    xexists,
    xgetsize,
    xglob,
    xgzip_open,
    xisdir,
    xisfile,
    xjoin,
    xlistdir,
    xnumpy_load,
    xopen,
    xpandas_read_csv,
    xpandas_read_excel,
    xPath,
    xrelpath,
    xsio_loadmat,
    xsplit,
    xsplitext,
    xwalk,
    xxml_dom_minidom_parse,
)
from .utils.logging import get_logger
from .utils.patching import patch_submodule
from .utils.py_utils import get_imports


lowercase__			=	get_logger(__name__)


if TYPE_CHECKING:
       from .builder import DatasetBuilder


def 				UpperCamelCase	(  snake_case	,     snake_case = None   ):




   '''simple docstring'''

   SCREAMING_SNAKE_CASE__			:				List[Any]       						= importlib.import_module(snake_case   )

   # TODO(QL): always update the module to add subsequent new authentication
   if hasattr(snake_case	,     '''_patched_for_streaming'''   ) and module._patched_for_streaming:
      return

   def wrap_auth(snake_case   ):
      @wraps(snake_case   )
      def wrapper(*snake_case	,     **snake_case   ):
         return function(*snake_case	,     use_auth_token=snake_case	,     **snake_case   )

      SCREAMING_SNAKE_CASE__			:				Any       						= '''wrap_auth'''
      return wrapper

   # open files in a streaming fashion
   patch_submodule(snake_case	,     '''open'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''os.listdir'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''os.walk'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''glob.glob'''	,     wrap_auth(snake_case   )   ).start()
   # allow to navigate in remote zip files
   patch_submodule(snake_case	,     '''os.path.join'''	,     snake_case   ).start()
   patch_submodule(snake_case	,     '''os.path.dirname'''	,     snake_case   ).start()
   patch_submodule(snake_case	,     '''os.path.basename'''	,     snake_case   ).start()
   patch_submodule(snake_case	,     '''os.path.relpath'''	,     snake_case   ).start()
   patch_submodule(snake_case	,     '''os.path.split'''	,     snake_case   ).start()
   patch_submodule(snake_case	,     '''os.path.splitext'''	,     snake_case   ).start()
   # allow checks on paths
   patch_submodule(snake_case	,     '''os.path.exists'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''os.path.isdir'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''os.path.isfile'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''os.path.getsize'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''pathlib.Path'''	,     snake_case   ).start()
   # file readers
   patch_submodule(snake_case	,     '''gzip.open'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''numpy.load'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''pandas.read_csv'''	,     wrap_auth(snake_case   )	,     attrs=['''__version__''']   ).start()
   patch_submodule(snake_case	,     '''pandas.read_excel'''	,     wrap_auth(snake_case   )	,     attrs=['''__version__''']   ).start()
   patch_submodule(snake_case	,     '''scipy.io.loadmat'''	,     wrap_auth(snake_case   )	,     attrs=['''__version__''']   ).start()
   patch_submodule(snake_case	,     '''xml.etree.ElementTree.parse'''	,     wrap_auth(snake_case   )   ).start()
   patch_submodule(snake_case	,     '''xml.dom.minidom.parse'''	,     wrap_auth(snake_case   )   ).start()
   SCREAMING_SNAKE_CASE__			:				Optional[int]       						= True







def 				UpperCamelCase	(  snake_case   ):




   '''simple docstring'''

   extend_module_for_streaming(builder.__module__	,     use_auth_token=builder.use_auth_token   )
   # if needed, we also have to extend additional internal imports (like wmt14 -> wmt_utils)
   if not builder.__module__.startswith('''datasets.'''   ):  # check that it's not a packaged builder like csv
      for imports in get_imports(inspect.getfile(builder.__class__   )   ):
         if imports[0] == "internal":
            SCREAMING_SNAKE_CASE__			:				Union[str, Any]       						= imports[1]
            SCREAMING_SNAKE_CASE__			:				int       						= '''.'''.join(builder.__module__.split('''.'''   )[:-1] + [internal_import_name]   )
            extend_module_for_streaming(snake_case	,     use_auth_token=builder.use_auth_token   )

    # builders can inherit from other builders that might use streaming functionality
    # (for example, ImageFolder and AudioFolder inherit from FolderBuilder which implements examples generation)
    # but these parents builders are not patched automatically as they are not instantiated, so we patch them here
   from .builder import DatasetBuilder

   SCREAMING_SNAKE_CASE__			:				str       						= [
       cls.__module__
       for cls in type(snake_case   ).__mro__[1:]  # make sure it's not the same module we've already patched
       if issubclass(cls	,     snake_case   ) and cls.__module__ != DatasetBuilder.__module__
   ]  # check it's not a standard builder from datasets.builder
   for module in parent_builder_modules:
      extend_module_for_streaming(snake_case	,     use_auth_token=builder.use_auth_token   )







'''simple docstring'''





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:
       lowercase__			=	False


if is_vision_available():
       from PIL import Image
else:






       class 						lowercase     :






             @staticmethod
             def    A__(*a__ ,  **a__					)       ->   Optional[Any]:
                pass






@is_pipeline_test
@require_vision
class 						lowercase     (   unittest.TestCase):
      __lowercase 		=  MODEL_FOR_VISION_2_SEQ_MAPPING
      __lowercase 		=  TF_MODEL_FOR_VISION_2_SEQ_MAPPING






      def    A__(self ,  a__ ,  a__ ,  a__					)       ->   Optional[Any]:
         SCREAMING_SNAKE_CASE__			:				Tuple       						= pipeline('''image-to-text''' ,  model=a__ ,  tokenizer=a__ ,  image_processor=a__					)
         SCREAMING_SNAKE_CASE__			:				List[Any]       						= [
             Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''					),
             '''./tests/fixtures/tests_samples/COCO/000000039769.png''',
         ]
         return pipe, examples






      def    A__(self ,  a__ ,  a__					)       ->   Any:
         SCREAMING_SNAKE_CASE__			:				Tuple       						= pipe(a__					)
         self.assertEqual(
             a__ ,  [
                 [{'''generated_text''': ANY(a__					)}],
                 [{'''generated_text''': ANY(a__					)}],
             ] ,  )






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

         SCREAMING_SNAKE_CASE__			:				str       						= pipe(a__					)
         self.assertEqual(
             a__ ,  [
                 {
                     '''generated_text''': '''growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'''
                 },
             ] ,  )

         SCREAMING_SNAKE_CASE__			:				Optional[Any]       						= pipe([image, image]					)
         self.assertEqual(
             a__ ,  [
                 [
                     {
                         '''generated_text''': '''growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'''
                     }
                 ],
                 [
                     {
                         '''generated_text''': '''growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'''
                     }
                 ],
             ] ,  )

         SCREAMING_SNAKE_CASE__			:				Union[str, Any]       						= pipe(a__ ,  max_new_tokens=1					)
         self.assertEqual(
             a__ ,  [{'''generated_text''': '''growth'''}] ,  )






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

         SCREAMING_SNAKE_CASE__			:				Optional[Any]       						= pipe(a__					)
         self.assertEqual(
             a__ ,  [
                 {
                     '''generated_text''': '''growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'''
                 },
             ] ,  )

         SCREAMING_SNAKE_CASE__			:				Any       						= pipe([image, image]					)
         self.assertEqual(
             a__ ,  [
                 [
                     {
                         '''generated_text''': '''growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'''
                     }
                 ],
                 [
                     {
                         '''generated_text''': '''growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO'''
                     }
                 ],
             ] ,  )






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

         SCREAMING_SNAKE_CASE__			:				List[Any]       						= pipe(a__ ,  prompt=a__					)
         self.assertTrue(outputs[0]['''generated_text'''].startswith(a__					)					)






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

         SCREAMING_SNAKE_CASE__			:				List[Any]       						= pipe(a__					)
         self.assertEqual(a__ ,  [{'''generated_text''': '''a cat laying on a blanket next to a cat laying on a bed '''}]					)

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

         SCREAMING_SNAKE_CASE__			:				Optional[int]       						= pipe(a__					)
         self.assertEqual(a__ ,  [{'''generated_text''': '''a pink pokemon pokemon with a blue shirt and a blue shirt'''}]					)






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

         SCREAMING_SNAKE_CASE__			:				str       						= pipe(a__					)
         self.assertEqual(a__ ,  [{'''generated_text''': '''a cartoon of a purple character.'''}]					)






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

         SCREAMING_SNAKE_CASE__			:				Union[str, Any]       						= '''a photography of'''

         SCREAMING_SNAKE_CASE__			:				Union[str, Any]       						= pipe(a__ ,  prompt=a__					)
         self.assertEqual(a__ ,  [{'''generated_text''': '''a photography of a volcano'''}]					)

         with self.assertRaises(a__					):
            SCREAMING_SNAKE_CASE__			:				Any       						= pipe([image, image] ,  prompt=[prompt, prompt]					)






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

         SCREAMING_SNAKE_CASE__			:				Optional[int]       						= '''a photo of a'''

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

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

         SCREAMING_SNAKE_CASE__			:				int       						= '''What does the label 15 represent? (1) lava (2) core (3) tunnel (4) ash cloud'''

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

         with self.assertRaises(a__					):
            SCREAMING_SNAKE_CASE__			:				Any       						= pipe([image, image] ,  prompt=[prompt, prompt]					)






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

         SCREAMING_SNAKE_CASE__			:				Dict       						= pipe(a__					)
         self.assertEqual(a__ ,  [{'''generated_text''': '''a cat laying on a blanket next to a cat laying on a bed '''}]					)

         SCREAMING_SNAKE_CASE__			:				Union[str, Any]       						= pipe([image, image]					)
         self.assertEqual(
             a__ ,  [
                 [{'''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 '''}],
             ] ,  )

