from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    is_valid_image,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
						import PIL


A_      =      logging.get_logger(__name__)






def        __UpperCAmelCase    (       UpperCAmelCase       )->       List[List[ImageInput]]:



			"""simple docstring"""







			if isinstance(UpperCAmelCase,		(list, tuple)       ) and isinstance(videos[0],		(list, tuple)       ) and is_valid_image(videos[0][0]       ):
						return videos

			elif isinstance(UpperCAmelCase,		(list, tuple)       ) and is_valid_image(videos[0]       ):
						return [videos]

			elif is_valid_image(UpperCAmelCase       ):
						return [[videos]]

			raise ValueError(f'Could not make batched video from {videos}'       )






class      __lowercase					(    _A    ):
						lowercase					 =     ['pixel_values']

						def __init__(	self				:      Union[str, Any]							,							__lowerCamelCase				:      bool = True							,							__lowerCamelCase				:      Dict[str, int] = None							,							__lowerCamelCase				:      PILImageResampling = PILImageResampling.BILINEAR							,							__lowerCamelCase				:      bool = True							,							__lowerCamelCase				:      Dict[str, int] = None							,							__lowerCamelCase				:      bool = True							,							__lowerCamelCase				:      Union[int, float] = 1 / 2_55							,							__lowerCamelCase				:      bool = True							,							__lowerCamelCase				:      Optional[Union[float, List[float]]] = None							,							__lowerCamelCase				:      Optional[Union[float, List[float]]] = None							,							**__lowerCamelCase				:      List[Any]							,							)							->		None:


									'''simple docstring'''

									super().__init__(**__lowerCamelCase   )
									lowercase							=		size if size is not None else {'''shortest_edge''': 2_24}
									lowercase							=		get_size_dict(__lowerCamelCase							,							default_to_square=__lowerCamelCase   )
									lowercase							=		crop_size if crop_size is not None else {'''height''': 2_24, '''width''': 2_24}
									lowercase							=		get_size_dict(__lowerCamelCase							,							param_name='''crop_size'''   )

									lowercase							=		do_resize
									lowercase							=		size
									lowercase							=		do_center_crop
									lowercase							=		crop_size
									lowercase							=		resample
									lowercase							=		do_rescale
									lowercase							=		rescale_factor
									lowercase							=		do_normalize
									lowercase							=		image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
									lowercase							=		image_std if image_std is not None else IMAGENET_STANDARD_STD

						def        __a		(	self				:      Optional[Any]							,							__lowerCamelCase				:      np.ndarray							,							__lowerCamelCase				:      Dict[str, int]							,							__lowerCamelCase				:      PILImageResampling = PILImageResampling.BILINEAR							,							__lowerCamelCase				:      Optional[Union[str, ChannelDimension]] = None							,							**__lowerCamelCase				:      List[Any]							,							)							->		np.ndarray:


									'''simple docstring'''

									lowercase							=		get_size_dict(__lowerCamelCase							,							default_to_square=__lowerCamelCase   )
									if "shortest_edge" in size:
												lowercase							=		get_resize_output_image_size(__lowerCamelCase							,							size['''shortest_edge''']							,							default_to_square=__lowerCamelCase   )
									elif "height" in size and "width" in size:
												lowercase							=		(size['''height'''], size['''width'''])
									else:
												raise ValueError(f'Size must have \'height\' and \'width\' or \'shortest_edge\' as keys. Got {size.keys()}'   )
									return resize(__lowerCamelCase							,							size=__lowerCamelCase							,							resample=__lowerCamelCase							,							data_format=__lowerCamelCase							,							**__lowerCamelCase   )

						def        __a		(	self				:      Any							,							__lowerCamelCase				:      np.ndarray							,							__lowerCamelCase				:      Dict[str, int]							,							__lowerCamelCase				:      Optional[Union[str, ChannelDimension]] = None							,							**__lowerCamelCase				:      Any							,							)							->		np.ndarray:


									'''simple docstring'''

									lowercase							=		get_size_dict(__lowerCamelCase   )
									if "height" not in size or "width" not in size:
												raise ValueError(f'Size must have \'height\' and \'width\' as keys. Got {size.keys()}'   )
									return center_crop(__lowerCamelCase							,							size=(size['''height'''], size['''width'''])							,							data_format=__lowerCamelCase							,							**__lowerCamelCase   )

						def        __a		(	self				:      Dict							,							__lowerCamelCase				:      np.ndarray							,							__lowerCamelCase				:      Union[int, float]							,							__lowerCamelCase				:      Optional[Union[str, ChannelDimension]] = None							,							**__lowerCamelCase				:      Tuple							,							)							->		int:


									'''simple docstring'''

									return rescale(__lowerCamelCase							,							scale=__lowerCamelCase							,							data_format=__lowerCamelCase							,							**__lowerCamelCase   )

						def        __a		(	self				:      Optional[Any]							,							__lowerCamelCase				:      np.ndarray							,							__lowerCamelCase				:      Union[float, List[float]]							,							__lowerCamelCase				:      Union[float, List[float]]							,							__lowerCamelCase				:      Optional[Union[str, ChannelDimension]] = None							,							**__lowerCamelCase				:      List[Any]							,							)							->		np.ndarray:


									'''simple docstring'''

									return normalize(__lowerCamelCase							,							mean=__lowerCamelCase							,							std=__lowerCamelCase							,							data_format=__lowerCamelCase							,							**__lowerCamelCase   )

						def        __a		(	self				:      Union[str, Any]							,							__lowerCamelCase				:      ImageInput							,							__lowerCamelCase				:      bool = None							,							__lowerCamelCase				:      Dict[str, int] = None							,							__lowerCamelCase				:      PILImageResampling = None							,							__lowerCamelCase				:      bool = None							,							__lowerCamelCase				:      Dict[str, int] = None							,							__lowerCamelCase				:      bool = None							,							__lowerCamelCase				:      float = None							,							__lowerCamelCase				:      bool = None							,							__lowerCamelCase				:      Optional[Union[float, List[float]]] = None							,							__lowerCamelCase				:      Optional[Union[float, List[float]]] = None							,							__lowerCamelCase				:      Optional[ChannelDimension] = ChannelDimension.FIRST							,							)							->		np.ndarray:


									'''simple docstring'''

									if do_resize and size is None or resample is None:
												raise ValueError('''Size and resample must be specified if do_resize is True.'''   )

									if do_center_crop and crop_size is None:
												raise ValueError('''Crop size must be specified if do_center_crop is True.'''   )

									if do_rescale and rescale_factor is None:
												raise ValueError('''Rescale factor must be specified if do_rescale is True.'''   )

									if do_normalize and (image_mean is None or image_std is None):
												raise ValueError('''Image mean and std must be specified if do_normalize is True.'''   )

									# All transformations expect numpy arrays.
									lowercase							=		to_numpy_array(__lowerCamelCase   )

									if do_resize:
												lowercase							=		self.resize(image=__lowerCamelCase							,							size=__lowerCamelCase							,							resample=__lowerCamelCase   )

									if do_center_crop:
												lowercase							=		self.center_crop(__lowerCamelCase							,							size=__lowerCamelCase   )

									if do_rescale:
												lowercase							=		self.rescale(image=__lowerCamelCase							,							scale=__lowerCamelCase   )

									if do_normalize:
												lowercase							=		self.normalize(image=__lowerCamelCase							,							mean=__lowerCamelCase							,							std=__lowerCamelCase   )

									lowercase							=		to_channel_dimension_format(__lowerCamelCase							,							__lowerCamelCase   )
									return image



						def        __a		(	self				:      Union[str, Any]							,							__lowerCamelCase				:      ImageInput							,							__lowerCamelCase				:      bool = None							,							__lowerCamelCase				:      Dict[str, int] = None							,							__lowerCamelCase				:      PILImageResampling = None							,							__lowerCamelCase				:      bool = None							,							__lowerCamelCase				:      Dict[str, int] = None							,							__lowerCamelCase				:      bool = None							,							__lowerCamelCase				:      float = None							,							__lowerCamelCase				:      bool = None							,							__lowerCamelCase				:      Optional[Union[float, List[float]]] = None							,							__lowerCamelCase				:      Optional[Union[float, List[float]]] = None							,							__lowerCamelCase				:      Optional[Union[str, TensorType]] = None							,							__lowerCamelCase				:      ChannelDimension = ChannelDimension.FIRST							,							**__lowerCamelCase				:      int							,							)							->		PIL.Image.Image:


									'''simple docstring'''

									lowercase							=		do_resize if do_resize is not None else self.do_resize
									lowercase							=		resample if resample is not None else self.resample
									lowercase							=		do_center_crop if do_center_crop is not None else self.do_center_crop
									lowercase							=		do_rescale if do_rescale is not None else self.do_rescale
									lowercase							=		rescale_factor if rescale_factor is not None else self.rescale_factor
									lowercase							=		do_normalize if do_normalize is not None else self.do_normalize
									lowercase							=		image_mean if image_mean is not None else self.image_mean
									lowercase							=		image_std if image_std is not None else self.image_std

									lowercase							=		size if size is not None else self.size
									lowercase							=		get_size_dict(__lowerCamelCase							,							default_to_square=__lowerCamelCase   )
									lowercase							=		crop_size if crop_size is not None else self.crop_size
									lowercase							=		get_size_dict(__lowerCamelCase							,							param_name='''crop_size'''   )

									if not valid_images(__lowerCamelCase   ):
												raise ValueError(
												    '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
												    '''torch.Tensor, tf.Tensor or jax.ndarray.'''   )

									lowercase							=		make_batched(__lowerCamelCase   )

									lowercase							=		[
									    [
									        self._preprocess_image(
									            image=__lowerCamelCase							,							do_resize=__lowerCamelCase							,							size=__lowerCamelCase							,							resample=__lowerCamelCase							,							do_center_crop=__lowerCamelCase							,							crop_size=__lowerCamelCase							,							do_rescale=__lowerCamelCase							,							rescale_factor=__lowerCamelCase							,							do_normalize=__lowerCamelCase							,							image_mean=__lowerCamelCase							,							image_std=__lowerCamelCase							,							data_format=__lowerCamelCase							,							)
									        for img in video
									    ]
									    for video in videos
									]

									lowercase							=		{'''pixel_values''': videos}
									return BatchFeature(data=__lowerCamelCase							,							tensor_type=__lowerCamelCase   )






import os
import tempfile
from functools import partial
from unittest import TestCase
from unittest.mock import patch

import datasets
import datasets.config

from .utils import require_beam






class      __lowercase					(    datasets.BeamBasedBuilder    ):

						def        __a		(	self				:      Tuple   )							->		Tuple:


									'''simple docstring'''

									return datasets.DatasetInfo(
									    features=datasets.Features({'''content''': datasets.Value('''string'''   )}   )							,							supervised_keys=__lowerCamelCase							,							)

						def        __a		(	self				:      Union[str, Any]							,							__lowerCamelCase				:      Union[str, Any]							,							__lowerCamelCase				:      Tuple   )							->		int:


									'''simple docstring'''

									return [datasets.SplitGenerator(name=datasets.Split.TRAIN							,							gen_kwargs={'''examples''': get_test_dummy_examples()}   )]



						def        __a		(	self				:      List[Any]							,							__lowerCamelCase				:      Any							,							__lowerCamelCase				:      Union[str, Any]   )							->		Optional[int]:


									'''simple docstring'''

									import apache_beam as beam

									return pipeline | "Load Examples" >> beam.Create(__lowerCamelCase   )






class      __lowercase					(    datasets.BeamBasedBuilder    ):

						def        __a		(	self				:      int   )							->		Any:


									'''simple docstring'''

									return datasets.DatasetInfo(
									    features=datasets.Features({'''a''': datasets.Sequence({'''b''': datasets.Value('''string'''   )}   )}   )							,							supervised_keys=__lowerCamelCase							,							)

						def        __a		(	self				:      Optional[int]							,							__lowerCamelCase				:      List[Any]							,							__lowerCamelCase				:      Optional[int]   )							->		int:


									'''simple docstring'''

									return [
									    datasets.SplitGenerator(name=datasets.Split.TRAIN							,							gen_kwargs={'''examples''': get_test_nested_examples()}   )
									]



						def        __a		(	self				:      Union[str, Any]							,							__lowerCamelCase				:      Tuple							,							__lowerCamelCase				:      Union[str, Any]   )							->		Tuple:


									'''simple docstring'''

									import apache_beam as beam

									return pipeline | "Load Examples" >> beam.Create(__lowerCamelCase   )






def        __UpperCAmelCase    (       )->       Dict:



			"""simple docstring"""







			return [(i, {"content": content}) for i, content in enumerate(['''foo''', '''bar''', '''foobar''']       )]






def        __UpperCAmelCase    (       )->       List[str]:



			"""simple docstring"""







			return [(i, {"a": {"b": [content]}}) for i, content in enumerate(['''foo''', '''bar''', '''foobar''']       )]






class      __lowercase					(    _A    ):

						@require_beam
						def        __a		(	self				:      Optional[int]   )							->		Optional[Any]:


									'''simple docstring'''

									lowercase							=		len(get_test_dummy_examples()   )
									with tempfile.TemporaryDirectory() as tmp_cache_dir:
												lowercase							=		DummyBeamDataset(cache_dir=__lowerCamelCase							,							beam_runner='''DirectRunner'''   )
												builder.download_and_prepare()
												self.assertTrue(
												    os.path.exists(
												        os.path.join(__lowerCamelCase							,							builder.name							,							'''default'''							,							'''0.0.0'''							,							f'{builder.name}-train.arrow'   )   )   )
												self.assertDictEqual(builder.info.features							,							datasets.Features({'''content''': datasets.Value('''string'''   )}   )   )
												lowercase							=		builder.as_dataset()
												self.assertEqual(dset['''train'''].num_rows							,							__lowerCamelCase   )
												self.assertEqual(dset['''train'''].info.splits['''train'''].num_examples							,							__lowerCamelCase   )
												self.assertDictEqual(dset['''train'''][0]							,							get_test_dummy_examples()[0][1]   )
												self.assertDictEqual(
												    dset['''train'''][expected_num_examples - 1]							,							get_test_dummy_examples()[expected_num_examples - 1][1]   )
												self.assertTrue(
												    os.path.exists(os.path.join(__lowerCamelCase							,							builder.name							,							'''default'''							,							'''0.0.0'''							,							'''dataset_info.json'''   )   )   )
												del dset

						@require_beam
						def        __a		(	self				:      Union[str, Any]   )							->		List[Any]:


									'''simple docstring'''

									import apache_beam as beam

									lowercase							=		beam.io.parquetio.WriteToParquet

									lowercase							=		len(get_test_dummy_examples()   )
									with tempfile.TemporaryDirectory() as tmp_cache_dir:
												lowercase							=		DummyBeamDataset(cache_dir=__lowerCamelCase							,							beam_runner='''DirectRunner'''   )
												with patch('''apache_beam.io.parquetio.WriteToParquet'''   ) as write_parquet_mock:
															lowercase							=		partial(__lowerCamelCase							,							num_shards=2   )
															builder.download_and_prepare()
												self.assertTrue(
												    os.path.exists(
												        os.path.join(
												            __lowerCamelCase							,							builder.name							,							'''default'''							,							'''0.0.0'''							,							f'{builder.name}-train-00000-of-00002.arrow'   )   )   )
												self.assertTrue(
												    os.path.exists(
												        os.path.join(
												            __lowerCamelCase							,							builder.name							,							'''default'''							,							'''0.0.0'''							,							f'{builder.name}-train-00000-of-00002.arrow'   )   )   )
												self.assertDictEqual(builder.info.features							,							datasets.Features({'''content''': datasets.Value('''string'''   )}   )   )
												lowercase							=		builder.as_dataset()
												self.assertEqual(dset['''train'''].num_rows							,							__lowerCamelCase   )
												self.assertEqual(dset['''train'''].info.splits['''train'''].num_examples							,							__lowerCamelCase   )
												# Order is not preserved when sharding, so we just check that all the elements are there
												self.assertListEqual(sorted(dset['''train''']['''content''']   )							,							sorted(['''foo''', '''bar''', '''foobar''']   )   )
												self.assertTrue(
												    os.path.exists(os.path.join(__lowerCamelCase							,							builder.name							,							'''default'''							,							'''0.0.0'''							,							'''dataset_info.json'''   )   )   )
												del dset

						@require_beam
						def        __a		(	self				:      int   )							->		int:


									'''simple docstring'''

									with tempfile.TemporaryDirectory() as tmp_cache_dir:
												lowercase							=		DummyBeamDataset(cache_dir=__lowerCamelCase   )
												self.assertRaises(datasets.builder.MissingBeamOptions							,							builder.download_and_prepare   )



						@require_beam
						def        __a		(	self				:      str   )							->		List[Any]:


									'''simple docstring'''

									lowercase							=		len(get_test_nested_examples()   )
									with tempfile.TemporaryDirectory() as tmp_cache_dir:
												lowercase							=		NestedBeamDataset(cache_dir=__lowerCamelCase							,							beam_runner='''DirectRunner'''   )
												builder.download_and_prepare()
												self.assertTrue(
												    os.path.exists(
												        os.path.join(__lowerCamelCase							,							builder.name							,							'''default'''							,							'''0.0.0'''							,							f'{builder.name}-train.arrow'   )   )   )
												self.assertDictEqual(
												    builder.info.features							,							datasets.Features({'''a''': datasets.Sequence({'''b''': datasets.Value('''string'''   )}   )}   )   )
												lowercase							=		builder.as_dataset()
												self.assertEqual(dset['''train'''].num_rows							,							__lowerCamelCase   )
												self.assertEqual(dset['''train'''].info.splits['''train'''].num_examples							,							__lowerCamelCase   )
												self.assertDictEqual(dset['''train'''][0]							,							get_test_nested_examples()[0][1]   )
												self.assertDictEqual(
												    dset['''train'''][expected_num_examples - 1]							,							get_test_nested_examples()[expected_num_examples - 1][1]   )
												self.assertTrue(
												    os.path.exists(os.path.join(__lowerCamelCase							,							builder.name							,							'''default'''							,							'''0.0.0'''							,							'''dataset_info.json'''   )   )   )
												del dset





