



import json
import os
import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import BertTokenizer, BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES
from transformers.testing_utils import require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
				from PIL import Image

				from transformers import AlignProcessor, EfficientNetImageProcessor


@require_vision
class       a							(			unittest.TestCase				):



		def 	__lowerCamelCase			(     self     :Dict  ):
						snake_case__	:      Tuple     		=	tempfile.mkdtemp()

						snake_case__	:      str     		=	[
						    '''[UNK]''',
						    '''[CLS]''',
						    '''[SEP]''',
						    '''[PAD]''',
						    '''[MASK]''',
						    '''want''',
						    '''##want''',
						    '''##ed''',
						    '''wa''',
						    '''un''',
						    '''runn''',
						    '''##ing''',
						    ''',''',
						    '''low''',
						    '''lowest''',
						]
						snake_case__	:      Any     		=	os.path.join(self.tmpdirname   ,VOCAB_FILES_NAMES['''vocab_file''']  )
						with open(self.vocab_file   ,'''w'''   ,encoding='''utf-8'''  ) as vocab_writer:
										vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]  )  )

						snake_case__	:      int     		=	{
						    '''do_resize''': True,
						    '''size''': 2_0,
						    '''do_center_crop''': True,
						    '''crop_size''': 1_8,
						    '''do_normalize''': True,
						    '''image_mean''': [0.4814_5466, 0.457_8275, 0.4082_1073],
						    '''image_std''': [0.2686_2954, 0.2613_0258, 0.2757_7711],
						}
						snake_case__	:      str     		=	os.path.join(self.tmpdirname   ,__lowercase  )
						with open(self.image_processor_file   ,'''w'''   ,encoding='''utf-8'''  ) as fp:
										json.dump(__lowercase   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :Optional[Any]   ,**__lowercase     :int  ):
						return BertTokenizer.from_pretrained(self.tmpdirname   ,**__lowercase  )



		def 	__lowerCamelCase			(     self     :Any   ,**__lowercase     :Dict  ):
						return BertTokenizerFast.from_pretrained(self.tmpdirname   ,**__lowercase  )



		def 	__lowerCamelCase			(     self     :Any   ,**__lowercase     :Tuple  ):
						return EfficientNetImageProcessor.from_pretrained(self.tmpdirname   ,**__lowercase  )



		def 	__lowerCamelCase			(     self     :Dict  ):
						shutil.rmtree(self.tmpdirname  )



		def 	__lowerCamelCase			(     self     :Optional[int]  ):
						snake_case__	:      Tuple     		=	[np.random.randint(2_5_5   ,size=(3, 3_0, 4_0_0)   ,dtype=np.uinta  )]
						snake_case__	:      Tuple     		=	[Image.fromarray(np.moveaxis(__lowercase   ,0   ,-1  )  ) for x in image_inputs]
						return image_inputs



		def 	__lowerCamelCase			(     self     :Any  ):
						snake_case__	:      Union[str, Any]     		=	self.get_tokenizer()
						snake_case__	:      int     		=	self.get_rust_tokenizer()
						snake_case__	:      Optional[int]     		=	self.get_image_processor()

						snake_case__	:      List[str]     		=	AlignProcessor(tokenizer=__lowercase   ,image_processor=__lowercase  )
						processor_slow.save_pretrained(self.tmpdirname  )
						snake_case__	:      List[str]     		=	AlignProcessor.from_pretrained(self.tmpdirname   ,use_fast=__lowercase  )

						snake_case__	:      str     		=	AlignProcessor(tokenizer=__lowercase   ,image_processor=__lowercase  )
						processor_fast.save_pretrained(self.tmpdirname  )
						snake_case__	:      int     		=	AlignProcessor.from_pretrained(self.tmpdirname  )

						self.assertEqual(processor_slow.tokenizer.get_vocab()   ,tokenizer_slow.get_vocab()  )
						self.assertEqual(processor_fast.tokenizer.get_vocab()   ,tokenizer_fast.get_vocab()  )
						self.assertEqual(tokenizer_slow.get_vocab()   ,tokenizer_fast.get_vocab()  )
						self.assertIsInstance(processor_slow.tokenizer   ,__lowercase  )
						self.assertIsInstance(processor_fast.tokenizer   ,__lowercase  )

						self.assertEqual(processor_slow.image_processor.to_json_string()   ,image_processor.to_json_string()  )
						self.assertEqual(processor_fast.image_processor.to_json_string()   ,image_processor.to_json_string()  )
						self.assertIsInstance(processor_slow.image_processor   ,__lowercase  )
						self.assertIsInstance(processor_fast.image_processor   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :Any  ):
						snake_case__	:      Dict     		=	AlignProcessor(tokenizer=self.get_tokenizer()   ,image_processor=self.get_image_processor()  )
						processor.save_pretrained(self.tmpdirname  )

						snake_case__	:      List[str]     		=	self.get_tokenizer(bos_token='''(BOS)'''   ,eos_token='''(EOS)'''  )
						snake_case__	:      Optional[Any]     		=	self.get_image_processor(do_normalize=__lowercase   ,padding_value=1.0  )

						snake_case__	:      str     		=	AlignProcessor.from_pretrained(
						    self.tmpdirname   ,bos_token='''(BOS)'''   ,eos_token='''(EOS)'''   ,do_normalize=__lowercase   ,padding_value=1.0  )

						self.assertEqual(processor.tokenizer.get_vocab()   ,tokenizer_add_kwargs.get_vocab()  )
						self.assertIsInstance(processor.tokenizer   ,__lowercase  )

						self.assertEqual(processor.image_processor.to_json_string()   ,image_processor_add_kwargs.to_json_string()  )
						self.assertIsInstance(processor.image_processor   ,__lowercase  )



		def 	__lowerCamelCase			(     self     :List[Any]  ):
						snake_case__	:      List[str]     		=	self.get_image_processor()
						snake_case__	:      int     		=	self.get_tokenizer()

						snake_case__	:      List[str]     		=	AlignProcessor(tokenizer=__lowercase   ,image_processor=__lowercase  )

						snake_case__	:      Optional[Any]     		=	self.prepare_image_inputs()

						snake_case__	:      str     		=	image_processor(__lowercase   ,return_tensors='''np'''  )
						snake_case__	:      List[Any]     		=	processor(images=__lowercase   ,return_tensors='''np'''  )

						for key in input_image_proc.keys():
										self.assertAlmostEqual(input_image_proc[key].sum()   ,input_processor[key].sum()   ,delta=1e-2  )



		def 	__lowerCamelCase			(     self     :List[Any]  ):
						snake_case__	:      List[str]     		=	self.get_image_processor()
						snake_case__	:      Optional[Any]     		=	self.get_tokenizer()

						snake_case__	:      int     		=	AlignProcessor(tokenizer=__lowercase   ,image_processor=__lowercase  )

						snake_case__	:      Any     		=	'''lower newer'''

						snake_case__	:      List[str]     		=	processor(text=__lowercase  )

						snake_case__	:      List[str]     		=	tokenizer(__lowercase   ,padding='''max_length'''   ,max_length=6_4  )

						for key in encoded_tok.keys():
										self.assertListEqual(encoded_tok[key]   ,encoded_processor[key]  )



		def 	__lowerCamelCase			(     self     :Optional[int]  ):
						snake_case__	:      Optional[Any]     		=	self.get_image_processor()
						snake_case__	:      Union[str, Any]     		=	self.get_tokenizer()

						snake_case__	:      Any     		=	AlignProcessor(tokenizer=__lowercase   ,image_processor=__lowercase  )

						snake_case__	:      Tuple     		=	'''lower newer'''
						snake_case__	:      List[str]     		=	self.prepare_image_inputs()

						snake_case__	:      Tuple     		=	processor(text=__lowercase   ,images=__lowercase  )

						self.assertListEqual(list(inputs.keys()  )   ,['''input_ids''', '''token_type_ids''', '''attention_mask''', '''pixel_values''']  )

						# test if it raises when no input is passed
						with pytest.raises(__lowercase  ):
										processor()



		def 	__lowerCamelCase			(     self     :Optional[Any]  ):
						snake_case__	:      Tuple     		=	self.get_image_processor()
						snake_case__	:      List[Any]     		=	self.get_tokenizer()

						snake_case__	:      List[Any]     		=	AlignProcessor(tokenizer=__lowercase   ,image_processor=__lowercase  )

						snake_case__	:      Optional[int]     		=	[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

						snake_case__	:      Tuple     		=	processor.batch_decode(__lowercase  )
						snake_case__	:      Tuple     		=	tokenizer.batch_decode(__lowercase  )

						self.assertListEqual(__lowercase   ,__lowercase  )






		def 	__lowerCamelCase			(     self     :int  ):
						snake_case__	:      Optional[int]     		=	self.get_image_processor()
						snake_case__	:      Tuple     		=	self.get_tokenizer()

						snake_case__	:      Optional[Any]     		=	AlignProcessor(tokenizer=__lowercase   ,image_processor=__lowercase  )

						snake_case__	:      Dict     		=	'''lower newer'''
						snake_case__	:      Any     		=	self.prepare_image_inputs()

						snake_case__	:      str     		=	processor(text=__lowercase   ,images=__lowercase  )

						self.assertListEqual(list(inputs.keys()  )   ,processor.model_input_names  )






from unittest.mock import patch

import pyspark

from datasets.packaged_modules.spark.spark import (
    Spark,
    SparkExamplesIterable,
    _generate_iterable_examples,
)

from ..utils import (
    require_dill_gt_0_3_2,
    require_not_windows,
)



def 			_lowerCAmelCase       (					__lowerCAmelCase  ,  __lowerCAmelCase      )   ->					Tuple:


				"""simple docstring"""

				snake_case__	:      Union[str, Any]     		=	[]
				for part_id in partition_order:
								snake_case__	:      Any     		=	df.where(f"""SPARK_PARTITION_ID() = {part_id}"""      ).collect()
								for row_idx, row in enumerate(__lowerCAmelCase      ):
												expected_row_ids_and_row_dicts.append((f"""{part_id}_{row_idx}""", row.asDict())      )
				return expected_row_ids_and_row_dicts



@require_not_windows
@require_dill_gt_0_3_2
def 			_lowerCAmelCase       (					)   ->					Tuple:


				"""simple docstring"""

				snake_case__	:      Any     		=	pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				snake_case__	:      Optional[Any]     		=	spark.range(100      ).repartition(1      )
				snake_case__	:      Optional[int]     		=	Spark(__lowerCAmelCase      )
				# The id ints will be converted to Pyarrow int64s, so each row will be 8 bytes. Setting a max_shard_size of 16 means
				# that each partition can hold 2 rows.
				spark_builder._repartition_df_if_needed(max_shard_size=16      )
				# Given that the dataframe has 100 rows and each partition has 2 rows, we expect 50 partitions.
				assert spark_builder.df.rdd.getNumPartitions() == 50



@require_not_windows
@require_dill_gt_0_3_2
def 			_lowerCAmelCase       (					)   ->					Optional[int]:


				"""simple docstring"""

				snake_case__	:      Any     		=	pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				snake_case__	:      Dict     		=	spark.range(10      ).repartition(2      )
				snake_case__	:      Any     		=	[1, 0]
				snake_case__	:      Tuple     		=	_generate_iterable_examples(__lowerCAmelCase  ,  __lowerCAmelCase      )  # Reverse the partitions.
				snake_case__	:      Any     		=	_get_expected_row_ids_and_row_dicts_for_partition_order(__lowerCAmelCase  ,  __lowerCAmelCase      )

				for i, (row_id, row_dict) in enumerate(generate_fn()      ):
								snake_case__		, snake_case__	:      Union[str, Any]     		=	expected_row_ids_and_row_dicts[i]
								assert row_id == expected_row_id
								assert row_dict == expected_row_dict



@require_not_windows
@require_dill_gt_0_3_2
def 			_lowerCAmelCase       (					)   ->					Any:


				"""simple docstring"""

				snake_case__	:      Any     		=	pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				snake_case__	:      List[Any]     		=	spark.range(10      ).repartition(1      )
				snake_case__	:      int     		=	SparkExamplesIterable(__lowerCAmelCase      )
				assert it.n_shards == 1
				for i, (row_id, row_dict) in enumerate(__lowerCAmelCase      ):
								assert row_id == f"""0_{i}"""
								assert row_dict == {"id": i}



@require_not_windows
@require_dill_gt_0_3_2
def 			_lowerCAmelCase       (					)   ->					Dict:


				"""simple docstring"""

				snake_case__	:      List[str]     		=	pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				snake_case__	:      Tuple     		=	spark.range(30      ).repartition(3      )
				# Mock the generator so that shuffle reverses the partition indices.
				with patch('''numpy.random.Generator'''      ) as generator_mock:
								snake_case__	:      Union[str, Any]     		=	lambda __lowerCAmelCase      : x.reverse()
								snake_case__	:      List[str]     		=	_get_expected_row_ids_and_row_dicts_for_partition_order(__lowerCAmelCase  ,  [2, 1, 0]      )

								snake_case__	:      List[str]     		=	SparkExamplesIterable(__lowerCAmelCase      ).shuffle_data_sources(__lowerCAmelCase      )
								assert shuffled_it.n_shards == 3
								for i, (row_id, row_dict) in enumerate(__lowerCAmelCase      ):
												snake_case__		, snake_case__	:      Optional[Any]     		=	expected_row_ids_and_row_dicts[i]
												assert row_id == expected_row_id
												assert row_dict == expected_row_dict



@require_not_windows
@require_dill_gt_0_3_2
def 			_lowerCAmelCase       (					)   ->					List[Any]:


				"""simple docstring"""

				snake_case__	:      Union[str, Any]     		=	pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				snake_case__	:      Dict     		=	spark.range(20      ).repartition(4      )

				# Partitions 0 and 2
				snake_case__	:      List[Any]     		=	SparkExamplesIterable(__lowerCAmelCase      ).shard_data_sources(worker_id=0  ,  num_workers=2      )
				assert shard_it_a.n_shards == 2
				snake_case__	:      int     		=	_get_expected_row_ids_and_row_dicts_for_partition_order(__lowerCAmelCase  ,  [0, 2]      )
				for i, (row_id, row_dict) in enumerate(__lowerCAmelCase      ):
								snake_case__		, snake_case__	:      Dict     		=	expected_row_ids_and_row_dicts_a[i]
								assert row_id == expected_row_id
								assert row_dict == expected_row_dict

				# Partitions 1 and 3
				snake_case__	:      List[str]     		=	SparkExamplesIterable(__lowerCAmelCase      ).shard_data_sources(worker_id=1  ,  num_workers=2      )
				assert shard_it_a.n_shards == 2
				snake_case__	:      List[Any]     		=	_get_expected_row_ids_and_row_dicts_for_partition_order(__lowerCAmelCase  ,  [1, 3]      )
				for i, (row_id, row_dict) in enumerate(__lowerCAmelCase      ):
								snake_case__		, snake_case__	:      Optional[int]     		=	expected_row_ids_and_row_dicts_a[i]
								assert row_id == expected_row_id
								assert row_dict == expected_row_dict





@require_not_windows
@require_dill_gt_0_3_2
def 			_lowerCAmelCase       (					)   ->					Dict:


				"""simple docstring"""

				snake_case__	:      int     		=	pyspark.sql.SparkSession.builder.master('''local[*]'''      ).appName('''pyspark'''      ).getOrCreate()
				snake_case__	:      Dict     		=	spark.range(100      ).repartition(1      )
				snake_case__	:      Tuple     		=	Spark(__lowerCAmelCase      )
				# Choose a small max_shard_size for maximum partitioning.
				spark_builder._repartition_df_if_needed(max_shard_size=1      )
				# The new number of partitions should not be greater than the number of rows.
				assert spark_builder.df.rdd.getNumPartitions() == 100

