"""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:
  __UpperCamelCase       :    Tuple         =		False


if is_vision_available():
  from PIL import Image
else:
  class   UpperCAmelCase_	:
      @staticmethod
      def 		_lowerCamelCase      (	*_lowercase			:       Tuple   ,						**_lowercase			:       Tuple		)						->    Optional[int]:
         pass
@is_pipeline_test
@require_vision
class   UpperCAmelCase_	(				unittest.TestCase   ):
    snake_case_											=      MODEL_FOR_VISION_2_SEQ_MAPPING
    snake_case_											=      TF_MODEL_FOR_VISION_2_SEQ_MAPPING
    def 		_lowerCamelCase      (	self			:       Optional[int]   ,						_lowercase			:       Optional[int]   ,						_lowercase			:       List[str]   ,						_lowercase			:       Any		)						->    Optional[Any]:
       _lowercase							     =       pipeline("image-to-text"   ,						model=lowerCamelCase__   ,						tokenizer=lowerCamelCase__   ,						image_processor=lowerCamelCase__		)
       _lowercase							     =       [
           Image.open("./tests/fixtures/tests_samples/COCO/000000039769.png"		),
           "./tests/fixtures/tests_samples/COCO/000000039769.png",
       ]
       return pipe, examples
    def 		_lowerCamelCase      (	self			:       Optional[int]   ,						_lowercase			:       List[str]   ,						_lowercase			:       str		)						->    Tuple:
       _lowercase							     =       pipe(lowerCamelCase__		)
       self.assertEqual(
           lowerCamelCase__   ,						[
               [{"generated_text": ANY(lowerCamelCase__		)}],
               [{"generated_text": ANY(lowerCamelCase__		)}],
           ]   ,						)
    @require_tf
    def 		_lowerCamelCase      (	self			:       Optional[int]		)						->    Optional[int]:
       _lowercase							     =       pipeline("image-to-text"   ,						model="hf-internal-testing/tiny-random-vit-gpt2"   ,						framework="tf"		)
       _lowercase							     =       "./tests/fixtures/tests_samples/COCO/000000039769.png"

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

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

       _lowercase							     =       pipe(lowerCamelCase__   ,						max_new_tokens=1		)
       self.assertEqual(
           lowerCamelCase__   ,						[{"generated_text": "growth"}]   ,						)
    @require_torch
    def 		_lowerCamelCase      (	self			:       str		)						->    Union[str, Any]:
       _lowercase							     =       pipeline("image-to-text"   ,						model="hf-internal-testing/tiny-random-vit-gpt2"		)
       _lowercase							     =       "./tests/fixtures/tests_samples/COCO/000000039769.png"

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

       _lowercase							     =       pipe([image, image]		)
       self.assertEqual(
           lowerCamelCase__   ,						[
               [
                   {
                       "generated_text": "growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"
                   }
               ],
               [
                   {
                       "generated_text": "growthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthgrowthGOGO"
                   }
               ],
           ]   ,						)
    @require_torch
    def 		_lowerCamelCase      (	self			:       Dict		)						->    Optional[Any]:
       _lowercase							     =       pipeline("image-to-text"   ,						model="hf-internal-testing/tiny-random-BlipForConditionalGeneration"		)
       _lowercase							     =       "./tests/fixtures/tests_samples/COCO/000000039769.png"
       _lowercase							     =       "a photo of"

       _lowercase							     =       pipe(lowerCamelCase__   ,						prompt=lowerCamelCase__		)
       self.assertTrue(outputs[0]["generated_text"].startswith(lowerCamelCase__		)		)
    @slow
    @require_torch
    def 		_lowerCamelCase      (	self			:       Union[str, Any]		)						->    Union[str, Any]:
       _lowercase							     =       pipeline("image-to-text"   ,						model="ydshieh/vit-gpt2-coco-en"		)
       _lowercase							     =       "./tests/fixtures/tests_samples/COCO/000000039769.png"

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

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

       _lowercase							     =       pipe(lowerCamelCase__		)
       self.assertEqual(lowerCamelCase__   ,						[{"generated_text": "a pink pokemon pokemon with a blue shirt and a blue shirt"}]		)
    @slow
    @require_torch
    def 		_lowerCamelCase      (	self			:       List[Any]		)						->    List[str]:
       _lowercase							     =       pipeline("image-to-text"   ,						model="microsoft/git-base-coco"		)
       _lowercase							     =       "https://huggingface.co/datasets/sayakpaul/sample-datasets/resolve/main/pokemon.png"
       _lowercase							     =       Image.open(requests.get(lowerCamelCase__   ,						stream=lowerCamelCase__		).raw		)

       _lowercase							     =       pipe(lowerCamelCase__		)
       self.assertEqual(lowerCamelCase__   ,						[{"generated_text": "a cartoon of a purple character."}]		)
    @slow
    @require_torch
    def 		_lowerCamelCase      (	self			:       str		)						->    int:
       _lowercase							     =       pipeline("image-to-text"   ,						model="Salesforce/blip-image-captioning-base"		)
       _lowercase							     =       "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"
       _lowercase							     =       Image.open(requests.get(lowerCamelCase__   ,						stream=lowerCamelCase__		).raw		)

       _lowercase							     =       "a photography of"

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

       with self.assertRaises(lowerCamelCase__		):
          _lowercase							     =       pipe([image, image]   ,						prompt=[prompt, prompt]		)
    @slow
    @require_torch
    def 		_lowerCamelCase      (	self			:       Union[str, Any]		)						->    Union[str, Any]:
       _lowercase							     =       pipeline("image-to-text"   ,						model="microsoft/git-base-coco"		)
       _lowercase							     =       "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"
       _lowercase							     =       Image.open(requests.get(lowerCamelCase__   ,						stream=lowerCamelCase__		).raw		)

       _lowercase							     =       "a photo of a"

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

       with self.assertRaises(lowerCamelCase__		):
          _lowercase							     =       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 		_lowerCamelCase      (	self			:       str		)						->    Any:
       _lowercase							     =       pipeline("image-to-text"   ,						model="google/pix2struct-ai2d-base"		)
       _lowercase							     =       "https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/transformers/tasks/ai2d-demo.jpg"
       _lowercase							     =       Image.open(requests.get(lowerCamelCase__   ,						stream=lowerCamelCase__		).raw		)

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

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

       with self.assertRaises(lowerCamelCase__		):
          _lowercase							     =       pipe([image, image]   ,						prompt=[prompt, prompt]		)




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

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

       _lowercase							     =       pipe([image, image]		)
       self.assertEqual(
           lowerCamelCase__   ,						[
               [{"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 "}],
           ]   ,						)



'''simple docstring'''


import os
import sys
from pathlib import Path

import pytest

from datasets import Dataset, IterableDataset
from datasets.distributed import split_dataset_by_node

from .utils import execute_subprocess_async, get_torch_dist_unique_port, require_torch






def 			__lowerCAmelCase			():
		_UpperCAmelCase		:							Optional[int]             =   Dataset.from_dict({"i": range(17 )} )
		_UpperCAmelCase		:							int             =   len(__lowerCAmelCase )
		_UpperCAmelCase		:							Dict             =   3
		_UpperCAmelCase		:							str             =   [
		    split_dataset_by_node(__lowerCAmelCase   ,			rank=__lowerCAmelCase   ,			world_size=__lowerCAmelCase ) for rank in range(__lowerCAmelCase )
		]
		assert sum(len(__lowerCAmelCase ) for ds in datasets_per_rank ) == full_size
		assert len({tuple(x.values() ) for ds in datasets_per_rank for x in ds} ) == full_size






def 			__lowerCAmelCase			():
		def gen():
				return ({"i": i} for i in range(17 ))

		_UpperCAmelCase		:							Any             =   3
		_UpperCAmelCase		:							Any             =   IterableDataset.from_generator(__lowerCAmelCase )
		_UpperCAmelCase		:							str             =   len(list(__lowerCAmelCase ) )
		_UpperCAmelCase		:							Tuple             =   [
		    split_dataset_by_node(__lowerCAmelCase   ,			rank=__lowerCAmelCase   ,			world_size=__lowerCAmelCase ) for rank in range(__lowerCAmelCase )
		]
		assert sum(len(list(__lowerCAmelCase ) ) for ds in datasets_per_rank ) == full_size
		assert len({tuple(x.values() ) for ds in datasets_per_rank for x in ds} ) == full_size






@pytest.mark.parametrize("shards_per_node"   ,			[1, 2, 3] )
def 			__lowerCAmelCase			(__lowerCAmelCase ):
		def gen(__lowerCAmelCase ):
				for shard in shards:
						yield from ({"i": i, "shard": shard} for i in range(17 ))

		_UpperCAmelCase		:							Tuple             =   3
		_UpperCAmelCase		:							Tuple             =   shards_per_node * world_size
		_UpperCAmelCase		:							Optional[int]             =   {"shards": [F"""shard_{shard_idx}.txt""" for shard_idx in range(__lowerCAmelCase )]}
		_UpperCAmelCase		:							int             =   IterableDataset.from_generator(__lowerCAmelCase   ,			gen_kwargs=__lowerCAmelCase )
		_UpperCAmelCase		:							Any             =   len(list(__lowerCAmelCase ) )
		assert full_ds.n_shards == world_size * shards_per_node
		_UpperCAmelCase		:							Optional[Any]             =   [
		    split_dataset_by_node(__lowerCAmelCase   ,			rank=__lowerCAmelCase   ,			world_size=__lowerCAmelCase ) for rank in range(__lowerCAmelCase )
		]
		assert [ds.n_shards for ds in datasets_per_rank] == [shards_per_node] * world_size
		assert sum(len(list(__lowerCAmelCase ) ) for ds in datasets_per_rank ) == full_size
		assert len({tuple(x.values() ) for ds in datasets_per_rank for x in ds} ) == full_size






def 			__lowerCAmelCase			():
		def gen():
				return ({"i": i} for i in range(17 ))

		_UpperCAmelCase		:							List[str]             =   2
		_UpperCAmelCase		:							int             =   IterableDataset.from_generator(__lowerCAmelCase )
		_UpperCAmelCase		:							Union[str, Any]             =   len(list(__lowerCAmelCase ) )

		_UpperCAmelCase		:							int             =   split_dataset_by_node(__lowerCAmelCase   ,			rank=0   ,			world_size=__lowerCAmelCase ).shuffle(seed=42 )
		assert len(list(__lowerCAmelCase ) ) == 1 + full_size // world_size
		with pytest.raises(__lowerCAmelCase ):
				split_dataset_by_node(__lowerCAmelCase   ,			rank=0   ,			world_size=__lowerCAmelCase ).shuffle()

		_UpperCAmelCase		:							Dict             =   split_dataset_by_node(full_ds.shuffle(seed=42 )   ,			rank=0   ,			world_size=__lowerCAmelCase )
		assert len(list(__lowerCAmelCase ) ) == 1 + full_size // world_size
		with pytest.raises(__lowerCAmelCase ):
				split_dataset_by_node(full_ds.shuffle()   ,			rank=0   ,			world_size=__lowerCAmelCase )






@pytest.mark.parametrize("streaming"   ,			[False, True] )
@require_torch
@pytest.mark.skipif(os.name == "nt"   ,			reason="execute_subprocess_async doesn't support windows" )
@pytest.mark.integration
def 			__lowerCAmelCase			(__lowerCAmelCase ):
		_UpperCAmelCase		:							Union[str, Any]             =   2
		_UpperCAmelCase		:							Dict             =   get_torch_dist_unique_port()
		_UpperCAmelCase		:							int             =   Path(__lowerCAmelCase ).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
		_UpperCAmelCase		:							List[Any]             =   F"""
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    """.split()
		_UpperCAmelCase		:							List[str]             =   F"""
        --streaming={streaming}
    """.split()
		_UpperCAmelCase		:							Tuple             =   [sys.executable] + distributed_args + args
		execute_subprocess_async(__lowerCAmelCase   ,			env=os.environ.copy() )






@pytest.mark.parametrize(
    "nproc_per_node, num_workers"   ,			[
        (2, 2),  # each node has 2 shards and each worker has 1 shards
        (3, 2),  # each node uses all the shards but skips examples, and each worker has 2 shards
    ]   ,			)
@require_torch
@pytest.mark.skipif(os.name == "nt"   ,			reason="execute_subprocess_async doesn't support windows" )
@pytest.mark.integration
def 			__lowerCAmelCase			(__lowerCAmelCase   ,			__lowerCAmelCase ):
		_UpperCAmelCase		:							Tuple             =   True
		_UpperCAmelCase		:							int             =   get_torch_dist_unique_port()
		_UpperCAmelCase		:							Optional[int]             =   Path(__lowerCAmelCase ).resolve().parent / "distributed_scripts" / "launch_torch_distributed.py"
		_UpperCAmelCase		:							List[str]             =   F"""
        -m torch.distributed.launch
        --nproc_per_node={nproc_per_node}
        --master_port={master_port}
        {test_script}
    """.split()
		_UpperCAmelCase		:							int             =   F"""
        --streaming={streaming}
        --num_workers={num_workers}
    """.split()
		_UpperCAmelCase		:							Optional[int]             =   [sys.executable] + distributed_args + args
		execute_subprocess_async(__lowerCAmelCase   ,			env=os.environ.copy() )



