

"""simple docstring"""





import os
from pathlib import Path
from unittest.mock import patch

import pytest
import zstandard as zstd

from datasets.download.download_config import DownloadConfig
from datasets.utils.file_utils import (
    OfflineModeIsEnabled,
    cached_path,
    fsspec_get,
    fsspec_head,
    ftp_get,
    ftp_head,
    get_from_cache,
    http_get,
    http_head,
)


_UpperCamelCase							=      '\\n    Text data.\n    Second line of data.'

_UpperCamelCase							=      'file'



@pytest.fixture(scope='''session'''   )
def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							int   ):


							'''simple docstring'''







							__lowerCamelCase      :      Dict         =tmp_path_factory.mktemp('''data'''   ) / (FILE_PATH + '.zstd')
							__lowerCamelCase      :      Optional[int]         =bytes(lowerCAmelCase__    ,						'''utf-8'''   )
							with zstd.open(lowerCAmelCase__    ,						'''wb'''   ) as f:
														f.write(lowerCAmelCase__   )
							return path



@pytest.fixture
def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							Optional[int]   ):


							'''simple docstring'''







							with open(os.path.join(tmpfs.local_root_dir    ,						lowerCAmelCase__   )    ,						'''w'''   ) as f:
														f.write(lowerCAmelCase__   )
							return FILE_PATH



@pytest.mark.parametrize('''compression_format'''    ,						['''gzip''', '''xz''', '''zstd''']   )
def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							Any    ,						SCREAMING_SNAKE_CASE       :							Union[str, Any]    ,						SCREAMING_SNAKE_CASE       :							Any    ,						SCREAMING_SNAKE_CASE       :							Any    ,						SCREAMING_SNAKE_CASE       :							Optional[int]    ,						SCREAMING_SNAKE_CASE       :							int   ):


							'''simple docstring'''







							__lowerCamelCase      :      int         ={'gzip': gz_file, 'xz': xz_file, 'zstd': zstd_path}
							__lowerCamelCase      :      Tuple         =input_paths[compression_format]
							__lowerCamelCase      :      int         =tmp_path / 'cache'
							__lowerCamelCase      :      List[str]         =DownloadConfig(cache_dir=lowerCAmelCase__    ,						extract_compressed_file=lowerCAmelCase__   )
							__lowerCamelCase      :      int         =cached_path(lowerCAmelCase__    ,						download_config=lowerCAmelCase__   )
							with open(lowerCAmelCase__   ) as f:
														__lowerCamelCase      :      Optional[Any]         =f.read()
							with open(lowerCAmelCase__   ) as f:
														__lowerCamelCase      :      Tuple         =f.read()
							assert extracted_file_content == expected_file_content



@pytest.mark.parametrize('''default_extracted'''    ,						[True, False]   )
@pytest.mark.parametrize('''default_cache_dir'''    ,						[True, False]   )
def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							int    ,						SCREAMING_SNAKE_CASE       :							Optional[int]    ,						SCREAMING_SNAKE_CASE       :							Dict    ,						SCREAMING_SNAKE_CASE       :							Tuple    ,						SCREAMING_SNAKE_CASE       :							Union[str, Any]   ):


							'''simple docstring'''







							__lowerCamelCase      :      List[str]         ='custom_cache'
							__lowerCamelCase      :      List[Any]         ='custom_extracted_dir'
							__lowerCamelCase      :      Optional[Any]         =tmp_path / 'custom_extracted_path'
							if default_extracted:
														__lowerCamelCase      :      Tuple         =('downloads' if default_cache_dir else custom_cache_dir, 'extracted')
							else:
														monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''    ,						lowerCAmelCase__   )
														monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''    ,						str(lowerCAmelCase__   )   )
														__lowerCamelCase      :      List[Any]         =custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

							__lowerCamelCase      :      Tuple         =xz_file
							__lowerCamelCase      :      List[str]         =(
							    DownloadConfig(extract_compressed_file=lowerCAmelCase__   )
							    if default_cache_dir
							    else DownloadConfig(cache_dir=tmp_path / custom_cache_dir    ,						extract_compressed_file=lowerCAmelCase__   )
							)
							__lowerCamelCase      :      List[str]         =cached_path(lowerCAmelCase__    ,						download_config=lowerCAmelCase__   )
							assert Path(lowerCAmelCase__   ).parent.parts[-2:] == expected



def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							Any   ):


							'''simple docstring'''







							__lowerCamelCase      :      Optional[int]         =str(Path(lowerCAmelCase__   ).resolve()   )
							assert cached_path(lowerCAmelCase__   ) == text_file
							# relative path
							__lowerCamelCase      :      int         =str(Path(lowerCAmelCase__   ).resolve().relative_to(Path(os.getcwd()   )   )   )
							assert cached_path(lowerCAmelCase__   ) == text_file



def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							Optional[Any]   ):


							'''simple docstring'''







							__lowerCamelCase      :      int         =str(tmp_path.resolve() / '''__missing_file__.txt'''   )
							with pytest.raises(lowerCAmelCase__   ):
														cached_path(lowerCAmelCase__   )
							# relative path
							__lowerCamelCase      :      int         ='./__missing_file__.txt'
							with pytest.raises(lowerCAmelCase__   ):
														cached_path(lowerCAmelCase__   )



def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							Optional[int]   ):


							'''simple docstring'''







							__lowerCamelCase      :      Union[str, Any]         =get_from_cache(F'tmp://{tmpfs_file}'   )
							with open(lowerCAmelCase__   ) as f:
														__lowerCamelCase      :      Optional[Any]         =f.read()
							assert output_file_content == FILE_CONTENT



@patch('''datasets.config.HF_DATASETS_OFFLINE'''    ,						lowerCAmelCase__   )
def 				lowerCAmelCase_			(      ):


							'''simple docstring'''







							with pytest.raises(lowerCAmelCase__   ):
														cached_path('''https://huggingface.co'''   )



@patch('''datasets.config.HF_DATASETS_OFFLINE'''    ,						lowerCAmelCase__   )
def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							List[str]   ):


							'''simple docstring'''







							__lowerCamelCase      :      Optional[int]         =tmp_path_factory.mktemp('''data'''   ) / 'file.html'
							with pytest.raises(lowerCAmelCase__   ):
														http_get('''https://huggingface.co'''    ,						temp_file=lowerCAmelCase__   )
							with pytest.raises(lowerCAmelCase__   ):
														http_head('''https://huggingface.co'''   )



@patch('''datasets.config.HF_DATASETS_OFFLINE'''    ,						lowerCAmelCase__   )
def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							Dict   ):


							'''simple docstring'''







							__lowerCamelCase      :      List[Any]         =tmp_path_factory.mktemp('''data'''   ) / 'file.html'
							with pytest.raises(lowerCAmelCase__   ):
														ftp_get('''ftp://huggingface.co'''    ,						temp_file=lowerCAmelCase__   )
							with pytest.raises(lowerCAmelCase__   ):
														ftp_head('''ftp://huggingface.co'''   )



@patch('''datasets.config.HF_DATASETS_OFFLINE'''    ,						lowerCAmelCase__   )
def 				lowerCAmelCase_			(      SCREAMING_SNAKE_CASE       :							Dict   ):


							'''simple docstring'''







							__lowerCamelCase      :      str         =tmp_path_factory.mktemp('''data'''   ) / 'file.html'
							with pytest.raises(lowerCAmelCase__   ):
														fsspec_get('''s3://huggingface.co'''    ,						temp_file=lowerCAmelCase__   )
							with pytest.raises(lowerCAmelCase__   ):
														fsspec_head('''s3://huggingface.co'''   )











"""simple docstring"""





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, normalize, rescale, resize, to_channel_dimension_format
from ...image_utils import (
    IMAGENET_STANDARD_MEAN,
    IMAGENET_STANDARD_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


if is_vision_available():
							import PIL


_UpperCamelCase							=      logging.get_logger(__name__)
class 		SCREAMING_SNAKE_CASE_							(    snake_case__   ):
				"""simple docstring"""
				__snake_case   :       List[str]	 =		["""pixel_values"""]




				def __init__(     self  :List[str]						,    __lowercase  :bool = True						,    __lowercase  :Dict[str, int] = None						,    __lowercase  :PILImageResampling = PIL.Image.BICUBIC						,    __lowercase  :bool = True						,    __lowercase  :Dict[str, int] = None						,    __lowercase  :Union[int, float] = 1 / 255						,    __lowercase  :bool = True						,    __lowercase  :bool = True						,    __lowercase  :Optional[Union[float, List[float]]] = None						,    __lowercase  :Optional[Union[float, List[float]]] = None						,    **__lowercase  :Union[str, Any]						,    ):
											super().__init__(**__lowercase			)
											__lowerCamelCase      :      Dict         =size if size is not None else {'''height''': 256, '''width''': 256}
											__lowerCamelCase      :      Union[str, Any]         =get_size_dict(__lowercase			)
											__lowerCamelCase      :      Tuple         =crop_size if crop_size is not None else {'''height''': 224, '''width''': 224}
											__lowerCamelCase      :      int         =get_size_dict(__lowercase						,    param_name='''crop_size'''			)

											__lowerCamelCase      :      Tuple         =do_resize
											__lowerCamelCase      :      int         =size
											__lowerCamelCase      :      Union[str, Any]         =resample
											__lowerCamelCase      :      Dict         =do_center_crop
											__lowerCamelCase      :      Any         =crop_size
											__lowerCamelCase      :      Dict         =do_rescale
											__lowerCamelCase      :      List[Any]         =rescale_factor
											__lowerCamelCase      :      Any         =do_normalize
											__lowerCamelCase      :      int         =image_mean if image_mean is not None else IMAGENET_STANDARD_MEAN
											__lowerCamelCase      :      List[str]         =image_std if image_std is not None else IMAGENET_STANDARD_STD




				def 					__lowercase  (     self  :Union[str, Any]						,    __lowercase  :np.ndarray						,    __lowercase  :Dict[str, int]						,    __lowercase  :PILImageResampling = PIL.Image.BICUBIC						,    __lowercase  :Optional[Union[str, ChannelDimension]] = None						,    **__lowercase  :Optional[int]						,    ):
											__lowerCamelCase      :      Optional[int]         =get_size_dict(__lowercase			)
											if "height" not in size or "width" not in size:
																		raise ValueError(f'The size dictionary must have keys \'height\' and \'width\'. Got {size.keys()}'			)
											return resize(
											    __lowercase						,    size=(size['''height'''], size['''width'''])						,    resample=__lowercase						,    data_format=__lowercase						,    **__lowercase			)




				def 					__lowercase  (     self  :Any						,    __lowercase  :np.ndarray						,    __lowercase  :Dict[str, int]						,    __lowercase  :Optional[Union[str, ChannelDimension]] = None						,    **__lowercase  :Optional[Any]						,    ):
											__lowerCamelCase      :      Dict         =get_size_dict(__lowercase			)
											if "height" not in size or "width" not in size:
																		raise ValueError(f'The size dictionary must have keys \'height\' and \'width\'. Got {size.keys()}'			)
											return center_crop(__lowercase						,    size=(size['''height'''], size['''width'''])						,    data_format=__lowercase						,    **__lowercase			)




				def 					__lowercase  (     self  :int						,    __lowercase  :np.ndarray						,    __lowercase  :Union[int, float]						,    __lowercase  :Optional[Union[str, ChannelDimension]] = None						,    **__lowercase  :List[Any]						,    ):
											return rescale(__lowercase						,    scale=__lowercase						,    data_format=__lowercase						,    **__lowercase			)




				def 					__lowercase  (     self  :Dict						,    __lowercase  :np.ndarray						,    __lowercase  :Union[float, List[float]]						,    __lowercase  :Union[float, List[float]]						,    __lowercase  :Optional[Union[str, ChannelDimension]] = None						,    **__lowercase  :int						,    ):
											return normalize(__lowercase						,    mean=__lowercase						,    std=__lowercase						,    data_format=__lowercase						,    **__lowercase			)




				def 					__lowercase  (     self  :Any						,    __lowercase  :ImageInput						,    __lowercase  :bool = None						,    __lowercase  :Dict[str, int] = None						,    __lowercase  :Optional[Any]=None						,    __lowercase  :bool = None						,    __lowercase  :Dict[str, int] = None						,    __lowercase  :bool = None						,    __lowercase  :float = None						,    __lowercase  :bool = None						,    __lowercase  :Optional[Union[float, List[float]]] = None						,    __lowercase  :Optional[Union[float, List[float]]] = None						,    __lowercase  :Optional[Union[str, TensorType]] = None						,    __lowercase  :ChannelDimension = ChannelDimension.FIRST						,    **__lowercase  :Any						,    ):
											__lowerCamelCase      :      Optional[Any]         =do_resize if do_resize is not None else self.do_resize
											__lowerCamelCase      :      List[Any]         =resample if resample is not None else self.resample
											__lowerCamelCase      :      Optional[int]         =do_center_crop if do_center_crop is not None else self.do_center_crop
											__lowerCamelCase      :      List[str]         =do_rescale if do_rescale is not None else self.do_rescale
											__lowerCamelCase      :      Optional[Any]         =rescale_factor if rescale_factor is not None else self.rescale_factor
											__lowerCamelCase      :      Optional[int]         =do_normalize if do_normalize is not None else self.do_normalize
											__lowerCamelCase      :      Dict         =image_mean if image_mean is not None else self.image_mean
											__lowerCamelCase      :      Union[str, Any]         =image_std if image_std is not None else self.image_std

											__lowerCamelCase      :      Dict         =size if size is not None else self.size
											__lowerCamelCase      :      int         =get_size_dict(__lowercase			)
											__lowerCamelCase      :      List[Any]         =crop_size if crop_size is not None else self.crop_size
											__lowerCamelCase      :      Optional[int]         =get_size_dict(__lowercase						,    param_name='''crop_size'''			)

											__lowerCamelCase      :      int         =make_list_of_images(__lowercase			)

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

											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.
											__lowerCamelCase      :      str         =[to_numpy_array(__lowercase			) for image in images]

											if do_resize:
																		__lowerCamelCase      :      Union[str, Any]         =[self.resize(image=__lowercase						,    size=__lowercase						,    resample=__lowercase			) for image in images]

											if do_center_crop:
																		__lowerCamelCase      :      Optional[Any]         =[self.center_crop(image=__lowercase						,    size=__lowercase			) for image in images]

											if do_rescale:
																		__lowerCamelCase      :      str         =[self.rescale(image=__lowercase						,    scale=__lowercase			) for image in images]

											if do_normalize:
																		__lowerCamelCase      :      int         =[self.normalize(image=__lowercase						,    mean=__lowercase						,    std=__lowercase			) for image in images]

											__lowerCamelCase      :      Tuple         =[to_channel_dimension_format(__lowercase						,    __lowercase			) for image in images]

											__lowerCamelCase      :      Any         ={'''pixel_values''': images}
											return BatchFeature(data=__lowercase						,    tensor_type=__lowercase			)




