



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 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Optional[Any]				)->str:
      _lowerCAmelCase 			=  tmp_path_factory.mktemp('''data'''				) / (FILE_PATH + '''.zstd''')
      _lowerCAmelCase 			=  bytes(_SCREAMING_SNAKE_CASE     ,				'''utf-8'''				)
      with zstd.open(_SCREAMING_SNAKE_CASE     ,				'''wb'''				) as f:
            f.write(_SCREAMING_SNAKE_CASE				)
      return path
@pytest.fixture
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Dict				)->int:
      with open(os.path.join(tmpfs.local_root_dir     ,				_SCREAMING_SNAKE_CASE				)     ,				'''w'''				) as f:
            f.write(_SCREAMING_SNAKE_CASE				)
      return FILE_PATH
@pytest.mark.parametrize('''compression_format'''     ,				['''gzip''', '''xz''', '''zstd''']				)
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Any     ,				_SCREAMING_SNAKE_CASE   :	Dict     ,				_SCREAMING_SNAKE_CASE   :	Tuple     ,				_SCREAMING_SNAKE_CASE   :	List[Any]     ,				_SCREAMING_SNAKE_CASE   :	Optional[int]     ,				_SCREAMING_SNAKE_CASE   :	Optional[int]				)->Optional[Any]:
      _lowerCAmelCase 			=  {'''gzip''': gz_file, '''xz''': xz_file, '''zstd''': zstd_path}
      _lowerCAmelCase 			=  input_paths[compression_format]
      _lowerCAmelCase 			=  tmp_path / '''cache'''
      _lowerCAmelCase 			=  DownloadConfig(cache_dir=_SCREAMING_SNAKE_CASE     ,				extract_compressed_file=_SCREAMING_SNAKE_CASE				)
      _lowerCAmelCase 			=  cached_path(_SCREAMING_SNAKE_CASE     ,				download_config=_SCREAMING_SNAKE_CASE				)
      with open(_SCREAMING_SNAKE_CASE				) as f:
            _lowerCAmelCase 			=  f.read()
      with open(_SCREAMING_SNAKE_CASE				) as f:
            _lowerCAmelCase 			=  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 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	str     ,				_SCREAMING_SNAKE_CASE   :	Tuple     ,				_SCREAMING_SNAKE_CASE   :	Any     ,				_SCREAMING_SNAKE_CASE   :	int     ,				_SCREAMING_SNAKE_CASE   :	Tuple				)->int:
      _lowerCAmelCase 			=  '''custom_cache'''
      _lowerCAmelCase 			=  '''custom_extracted_dir'''
      _lowerCAmelCase 			=  tmp_path / '''custom_extracted_path'''
      if default_extracted:
            _lowerCAmelCase 			=  ('''downloads''' if default_cache_dir else custom_cache_dir, '''extracted''')
      else:
            monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_DIR'''     ,				_SCREAMING_SNAKE_CASE				)
            monkeypatch.setattr('''datasets.config.EXTRACTED_DATASETS_PATH'''     ,				str(_SCREAMING_SNAKE_CASE				)				)
            _lowerCAmelCase 			=  custom_extracted_path.parts[-2:] if default_cache_dir else (custom_cache_dir, custom_extracted_dir)

      _lowerCAmelCase 			=  xz_file
      _lowerCAmelCase 			=  (
          DownloadConfig(extract_compressed_file=_SCREAMING_SNAKE_CASE				)
          if default_cache_dir
          else DownloadConfig(cache_dir=tmp_path / custom_cache_dir     ,				extract_compressed_file=_SCREAMING_SNAKE_CASE				)
      )
      _lowerCAmelCase 			=  cached_path(_SCREAMING_SNAKE_CASE     ,				download_config=_SCREAMING_SNAKE_CASE				)
      assert Path(_SCREAMING_SNAKE_CASE				).parent.parts[-2:] == expected
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Union[str, Any]				)->Union[str, Any]:
      # absolute path
      _lowerCAmelCase 			=  str(Path(_SCREAMING_SNAKE_CASE				).resolve()				)
      assert cached_path(_SCREAMING_SNAKE_CASE				) == text_file
      # relative path
      _lowerCAmelCase 			=  str(Path(_SCREAMING_SNAKE_CASE				).resolve().relative_to(Path(os.getcwd()				)				)				)
      assert cached_path(_SCREAMING_SNAKE_CASE				) == text_file
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Any				)->int:
      # absolute path
      _lowerCAmelCase 			=  str(tmp_path.resolve() / '''__missing_file__.txt'''				)
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            cached_path(_SCREAMING_SNAKE_CASE				)
      # relative path
      _lowerCAmelCase 			=  '''./__missing_file__.txt'''
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            cached_path(_SCREAMING_SNAKE_CASE				)
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Any				)->Tuple:
      _lowerCAmelCase 			=  get_from_cache(f'''tmp://{tmpfs_file}'''				)
      with open(_SCREAMING_SNAKE_CASE				) as f:
            _lowerCAmelCase 			=  f.read()
      assert output_file_content == FILE_CONTENT
@patch('''datasets.config.HF_DATASETS_OFFLINE'''     ,				_SCREAMING_SNAKE_CASE				)
def 							UpperCAmelCase__		(				)->Optional[int]:
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            cached_path('''https://huggingface.co'''				)
@patch('''datasets.config.HF_DATASETS_OFFLINE'''     ,				_SCREAMING_SNAKE_CASE				)
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	int				)->Optional[int]:
      _lowerCAmelCase 			=  tmp_path_factory.mktemp('''data'''				) / '''file.html'''
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            http_get('''https://huggingface.co'''     ,				temp_file=_SCREAMING_SNAKE_CASE				)
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            http_head('''https://huggingface.co'''				)
@patch('''datasets.config.HF_DATASETS_OFFLINE'''     ,				_SCREAMING_SNAKE_CASE				)
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	Any				)->List[Any]:
      _lowerCAmelCase 			=  tmp_path_factory.mktemp('''data'''				) / '''file.html'''
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            ftp_get('''ftp://huggingface.co'''     ,				temp_file=_SCREAMING_SNAKE_CASE				)
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            ftp_head('''ftp://huggingface.co'''				)


@patch('''datasets.config.HF_DATASETS_OFFLINE'''     ,				_SCREAMING_SNAKE_CASE				)
def 							UpperCAmelCase__		(				_SCREAMING_SNAKE_CASE   :	List[str]				)->Any:
      _lowerCAmelCase 			=  tmp_path_factory.mktemp('''data'''				) / '''file.html'''
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            fsspec_get('''s3://huggingface.co'''     ,				temp_file=_SCREAMING_SNAKE_CASE				)
      with pytest.raises(_SCREAMING_SNAKE_CASE				):
            fsspec_head('''s3://huggingface.co'''				)




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, prepare_video_inputs


if is_torch_available():
    import torch

if is_vision_available():
    from PIL import Image

    from transformers import VivitImageProcessor





class 		UpperCAmelCase						(						unittest.TestCase				):





   def __init__(	self       ,     _lowerCAmelCase       ,     _lowerCAmelCase=7       ,     _lowerCAmelCase=3       ,     _lowerCAmelCase=10       ,     _lowerCAmelCase=18       ,     _lowerCAmelCase=30       ,     _lowerCAmelCase=400       ,     _lowerCAmelCase=True       ,     _lowerCAmelCase=None       ,     _lowerCAmelCase=True       ,     _lowerCAmelCase=[0.5, 0.5, 0.5]       ,     _lowerCAmelCase=[0.5, 0.5, 0.5]       ,     _lowerCAmelCase=None       ,     ):
         _lowerCAmelCase 			=  size if size is not None else {'''shortest_edge''': 18}
         _lowerCAmelCase 			=  crop_size if crop_size is not None else {'''height''': 18, '''width''': 18}

         _lowerCAmelCase 			=  parent
         _lowerCAmelCase 			=  batch_size
         _lowerCAmelCase 			=  num_channels
         _lowerCAmelCase 			=  num_frames
         _lowerCAmelCase 			=  image_size
         _lowerCAmelCase 			=  min_resolution
         _lowerCAmelCase 			=  max_resolution
         _lowerCAmelCase 			=  do_resize
         _lowerCAmelCase 			=  size
         _lowerCAmelCase 			=  do_normalize
         _lowerCAmelCase 			=  image_mean
         _lowerCAmelCase 			=  image_std
         _lowerCAmelCase 			=  crop_size





   def 	__lowerCAmelCase							(	self			):
         return {
             "image_mean": self.image_mean,
             "image_std": self.image_std,
             "do_normalize": self.do_normalize,
             "do_resize": self.do_resize,
             "size": self.size,
             "crop_size": self.crop_size,
         }






@require_torch
@require_vision
class 		UpperCAmelCase						(						snake_case_       ,unittest.TestCase				):
   SCREAMING_SNAKE_CASE__							=     VivitImageProcessor if is_vision_available() else None





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  VivitImageProcessingTester(self			)





   @property
   def 	__lowerCAmelCase							(	self			):
         return self.image_processor_tester.prepare_image_processor_dict()





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''image_mean'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''image_std'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_normalize'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_resize'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''do_center_crop'''			)			)
         self.assertTrue(hasattr(_lowerCAmelCase       ,     '''size'''			)			)





   def 	__lowerCAmelCase							(	self			):
         _lowerCAmelCase 			=  self.image_processing_class.from_dict(self.image_processor_dict			)
         self.assertEqual(image_processor.size       ,     {'''shortest_edge''': 18}			)
         self.assertEqual(image_processor.crop_size       ,     {'''height''': 18, '''width''': 18}			)

         _lowerCAmelCase 			=  self.image_processing_class.from_dict(self.image_processor_dict       ,     size=42       ,     crop_size=84			)
         self.assertEqual(image_processor.size       ,     {'''shortest_edge''': 42}			)
         self.assertEqual(image_processor.crop_size       ,     {'''height''': 84, '''width''': 84}			)





   def 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random PIL videos
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     Image.Image			)

         # Test not batched input
         _lowerCAmelCase 			=  image_processing(video_inputs[0]       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 1,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )

         # Test batched
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )





   def 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random numpy tensors
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase       ,     numpify=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     np.ndarray			)

         # Test not batched input
         _lowerCAmelCase 			=  image_processing(video_inputs[0]       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 1,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )

         # Test batched
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )





   def 	__lowerCAmelCase							(	self			):
         # Initialize image_processing
         _lowerCAmelCase 			=  self.image_processing_class(**self.image_processor_dict			)
         # create random PyTorch tensors
         _lowerCAmelCase 			=  prepare_video_inputs(self.image_processor_tester       ,     equal_resolution=_lowerCAmelCase       ,     torchify=_lowerCAmelCase			)
         for video in video_inputs:
               self.assertIsInstance(_lowerCAmelCase       ,     _lowerCAmelCase			)
               self.assertIsInstance(video[0]       ,     torch.Tensor			)

         # Test not batched input
         _lowerCAmelCase 			=  image_processing(video_inputs[0]       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 1,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )

         # Test batched
         _lowerCAmelCase 			=  image_processing(_lowerCAmelCase       ,     return_tensors='''pt'''			).pixel_values
         self.assertEqual(
             encoded_videos.shape       ,     (
                 self.image_processor_tester.batch_size,
                 self.image_processor_tester.num_frames,
                 self.image_processor_tester.num_channels,
                 self.image_processor_tester.crop_size['''height'''],
                 self.image_processor_tester.crop_size['''width'''],
             )       ,     )