
"""simple docstring"""




import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
	import torch

if is_vision_available():
	from PIL import Image

	from transformers import SwinaSRImageProcessor
	from transformers.image_transforms import get_image_size




class      UpperCamelCase	(       unittest.TestCase      ):

				"""simple docstring"""



				def __init__(				self		,	lowercase__		,	lowercase__=7		,	lowercase__=3		,	lowercase__=18		,	lowercase__=30		,	lowercase__=400		,	lowercase__=True		,	lowercase__=1 / 255		,	lowercase__=True		,	lowercase__=8		,	)    ->    Any:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	parent
									SCREAMING_SNAKE_CASE				=	batch_size
									SCREAMING_SNAKE_CASE				=	num_channels
									SCREAMING_SNAKE_CASE				=	image_size
									SCREAMING_SNAKE_CASE				=	min_resolution
									SCREAMING_SNAKE_CASE				=	max_resolution
									SCREAMING_SNAKE_CASE				=	do_rescale
									SCREAMING_SNAKE_CASE				=	rescale_factor
									SCREAMING_SNAKE_CASE				=	do_pad
									SCREAMING_SNAKE_CASE				=	pad_size


				def 							A					(				self			)    ->    Union[str, Any]:
									"""simple docstring"""







									return {
									    "do_rescale": self.do_rescale,
									    "rescale_factor": self.rescale_factor,
									    "do_pad": self.do_pad,
									    "pad_size": self.pad_size,
									}




				def 							A					(				self		,	lowercase__=False		,	lowercase__=False		,	lowercase__=False			)    ->    Tuple:
									"""simple docstring"""







									assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

									if equal_resolution:
														SCREAMING_SNAKE_CASE				=	[]
														for i in range(self.batch_size			):
																			image_inputs.append(
																			    np.random.randint(
																			        255		,	size=(self.num_channels, self.max_resolution, self.max_resolution)		,	dtype=np.uinta			)			)
									else:
														SCREAMING_SNAKE_CASE				=	[]
														for i in range(self.batch_size			):
																			SCREAMING_SNAKE_CASE	,    SCREAMING_SNAKE_CASE				=	np.random.choice(np.arange(self.min_resolution		,	self.max_resolution			)		,	2			)
																			image_inputs.append(np.random.randint(255		,	size=(self.num_channels, width, height)		,	dtype=np.uinta			)			)

									if not numpify and not torchify:
														# PIL expects the channel dimension as last dimension
														SCREAMING_SNAKE_CASE				=	[Image.fromarray(np.moveaxis(lowercase__		,	0		,	-1			)			) for x in image_inputs]

									if torchify:
														SCREAMING_SNAKE_CASE				=	[torch.from_numpy(lowercase__			) for x in image_inputs]

									return image_inputs






@require_torch
@require_vision
class      UpperCamelCase	(       __magic_name__  ,      unittest.TestCase      ):

				"""simple docstring"""



				UpperCAmelCase_				:      int						  =				SwinaSRImageProcessor if is_vision_available() else None


				def 							A					(				self			)    ->    Union[str, Any]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	SwinaSRImageProcessingTester(self			)


				@property
				def 							A					(				self			)    ->    Dict:
									"""simple docstring"""







									return self.image_processor_tester.prepare_image_processor_dict()


				def 							A					(				self			)    ->    List[str]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	self.image_processing_class(**self.image_processor_dict			)
									self.assertTrue(hasattr(lowercase__		,	'do_rescale'			)			)
									self.assertTrue(hasattr(lowercase__		,	'rescale_factor'			)			)
									self.assertTrue(hasattr(lowercase__		,	'do_pad'			)			)
									self.assertTrue(hasattr(lowercase__		,	'pad_size'			)			)


				def 							A					(				self			)    ->    Dict:
									"""simple docstring"""







									pass


				def 							A					(				self		,	lowercase__			)    ->    Optional[Any]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE	,    SCREAMING_SNAKE_CASE				=	get_image_size(lowercase__			)
									SCREAMING_SNAKE_CASE				=	self.image_processor_tester.pad_size

									SCREAMING_SNAKE_CASE				=	(old_height // size + 1) * size - old_height
									SCREAMING_SNAKE_CASE				=	(old_width // size + 1) * size - old_width
									return old_height + pad_height, old_width + pad_width


				def 							A					(				self			)    ->    Tuple:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	self.image_processing_class(**self.image_processor_dict			)
									# create random PIL images
									SCREAMING_SNAKE_CASE				=	self.image_processor_tester.prepare_inputs(equal_resolution=lowercase__			)
									for image in image_inputs:
														self.assertIsInstance(lowercase__		,	Image.Image			)

									# Test not batched input
									SCREAMING_SNAKE_CASE				=	image_processor(image_inputs[0]		,	return_tensors='pt'			).pixel_values
									SCREAMING_SNAKE_CASE	,    SCREAMING_SNAKE_CASE				=	self.calculate_expected_size(np.array(image_inputs[0]			)			)
									self.assertEqual(
									    encoded_images.shape		,	(
									        1,
									        self.image_processor_tester.num_channels,
									        expected_height,
									        expected_width,
									    )		,	)


				def 							A					(				self			)    ->    List[str]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	self.image_processing_class(**self.image_processor_dict			)
									# create random numpy tensors
									SCREAMING_SNAKE_CASE				=	self.image_processor_tester.prepare_inputs(equal_resolution=lowercase__		,	numpify=lowercase__			)
									for image in image_inputs:
														self.assertIsInstance(lowercase__		,	np.ndarray			)

									# Test not batched input
									SCREAMING_SNAKE_CASE				=	image_processor(image_inputs[0]		,	return_tensors='pt'			).pixel_values
									SCREAMING_SNAKE_CASE	,    SCREAMING_SNAKE_CASE				=	self.calculate_expected_size(image_inputs[0]			)
									self.assertEqual(
									    encoded_images.shape		,	(
									        1,
									        self.image_processor_tester.num_channels,
									        expected_height,
									        expected_width,
									    )		,	)




				def 							A					(				self			)    ->    int:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	self.image_processing_class(**self.image_processor_dict			)
									# create random PyTorch tensors
									SCREAMING_SNAKE_CASE				=	self.image_processor_tester.prepare_inputs(equal_resolution=lowercase__		,	torchify=lowercase__			)
									for image in image_inputs:
														self.assertIsInstance(lowercase__		,	torch.Tensor			)

									# Test not batched input
									SCREAMING_SNAKE_CASE				=	image_processor(image_inputs[0]		,	return_tensors='pt'			).pixel_values
									SCREAMING_SNAKE_CASE	,    SCREAMING_SNAKE_CASE				=	self.calculate_expected_size(image_inputs[0]			)
									self.assertEqual(
									    encoded_images.shape		,	(
									        1,
									        self.image_processor_tester.num_channels,
									        expected_height,
									        expected_width,
									    )		,	)


"""simple docstring"""




import json
import os
from dataclasses import dataclass

import numpy as np
import pyarrow as pa

import datasets
from utils import get_duration


snake_case    =		1_0_0_0_0_0_0_0_0_0_0_0
snake_case    =		1_0_0_0_0

snake_case   ,			snake_case    =		os.path.split(__file__)
snake_case    =		os.path.join(RESULTS_BASEPATH, 'results', RESULTS_FILENAME.replace('.py', '.json'))

def   UpperCamelCase_  (		SCREAMING_SNAKE_CASE_,     SCREAMING_SNAKE_CASE_			):
					SCREAMING_SNAKE_CASE				=	pa.Table.from_pydict({'col': [0] * chunk_size}			)
					SCREAMING_SNAKE_CASE				=	pa.concat_tables([table] * (num_examples // chunk_size)			)
					return datasets.Dataset(SCREAMING_SNAKE_CASE_,     fingerprint='table_100B'			)




@dataclass
class      UpperCamelCase	:

				"""simple docstring"""



				UpperCAmelCase_				:      int
				UpperCAmelCase_				:      int
				UpperCAmelCase_				:      int
				UpperCAmelCase_				:      int


				def 							A					(				self			)    ->    Optional[Any]:
									"""simple docstring"""







									SCREAMING_SNAKE_CASE				=	np.random.default_rng(self.seed			)
									SCREAMING_SNAKE_CASE				=	rng.integers(low=self.low		,	high=self.high		,	size=self.size			).tolist()


				def __iter__(				self			)    ->    Union[str, Any]:
									"""simple docstring"""







									return iter(self._sampled_values			)




				def __len__(				self			)    ->    Optional[Any]:
									"""simple docstring"""







									return self.size






@get_duration
def   UpperCamelCase_  (		SCREAMING_SNAKE_CASE_			):
					SCREAMING_SNAKE_CASE				=	dataset[0]

@get_duration
def   UpperCamelCase_  (		SCREAMING_SNAKE_CASE_			):
					SCREAMING_SNAKE_CASE				=	dataset[-1]

@get_duration
def   UpperCamelCase_  (		SCREAMING_SNAKE_CASE_			):
					SCREAMING_SNAKE_CASE				=	dataset[range(len(SCREAMING_SNAKE_CASE_			) // 2,     len(SCREAMING_SNAKE_CASE_			) // 2 + 1_0_2_4			)]

@get_duration
def   UpperCamelCase_  (		SCREAMING_SNAKE_CASE_			):
					SCREAMING_SNAKE_CASE				=	dataset[RandIter(0,     len(SCREAMING_SNAKE_CASE_			),     1_0_2_4,     seed=4_2			)]

def   UpperCamelCase_  (		):
					SCREAMING_SNAKE_CASE				=	{'num examples': SPEED_TEST_N_EXAMPLES}
					SCREAMING_SNAKE_CASE				=	(get_first_row, get_last_row, get_batch_of_aaaa_rows, get_batch_of_aaaa_random_rows)
					print('generating dataset'			)
					SCREAMING_SNAKE_CASE				=	generate_100B_dataset(num_examples=SCREAMING_SNAKE_CASE_,     chunk_size=SCREAMING_SNAKE_CASE_			)
					print('Functions'			)
					for func in functions:
										print(func.__name__			)
										SCREAMING_SNAKE_CASE				=	func(SCREAMING_SNAKE_CASE_			)

					with open(SCREAMING_SNAKE_CASE_,     'wb'			) as f:
										f.write(json.dumps(SCREAMING_SNAKE_CASE_			).encode('utf-8'			)			)


if __name__ == "__main__":  # useful to run the profiler
	benchmark_table_100B()
