



'''simple docstring'''





import os
import re
import urllib.parse
from pathlib import Path
from typing import Callable, List, Optional, Union
from zipfile import ZipFile

from ..utils.file_utils import cached_path, hf_github_url
from ..utils.logging import get_logger
from ..utils.version import Version


A					:       Tuple									=     get_logger(__name__)





class 				lowerCAmelCase_						:
    __UpperCAmelCase				=			'dummy_data'
    __UpperCAmelCase				=			'datasets'
    __UpperCAmelCase				=			False





    def __init__(							self							:							str,							_snake_case							:							Union[str, Any],							_snake_case							:							Any,							_snake_case							:							Tuple,							_snake_case							:							int = None,							_snake_case							:							Optional[int] = False,							_snake_case							:							List[str] = True,							_snake_case							:							List[str] = None,							):



       '''simple docstring'''







       snake_case       :				Optional[int]		    =0
       snake_case       :				int		    =dataset_name
       snake_case       :				Tuple		    =cache_dir
       snake_case       :				Union[str, Any]		    =use_local_dummy_data
       snake_case       :				List[Any]		    =config
       # download_callbacks take a single url as input
       snake_case       :				List[Callable]		    =download_callbacks or []
       # if False, it doesn't load existing files and it returns the paths of the dummy files relative
       # to the dummy_data zip file root
       snake_case       :				int		    =load_existing_dummy_data

       # TODO(PVP, QL) might need to make this more general
       snake_case       :				Union[str, Any]		    =str(_snake_case      )
       # to be downloaded
       snake_case       :				str		    =None
       snake_case       :				Dict		    =None





    @property
    def 	__snake_case  (							self							:							Dict      ):



       '''simple docstring'''







       if self._dummy_file is None:
          snake_case       :				Optional[Any]		    =self.download_dummy_data()
       return self._dummy_file





    @property
    def 	__snake_case  (							self							:							List[Any]      ):



       '''simple docstring'''







       if self.config is not None:
          # structure is dummy / config_name / version_name
          return os.path.join('''dummy''',							self.config.name,							self.version_name      )
       # structure is dummy / version_name
       return os.path.join('''dummy''',							self.version_name      )





    @property
    def 	__snake_case  (							self							:							str      ):



       '''simple docstring'''







       return os.path.join(self.dummy_data_folder,							'''dummy_data.zip'''      )





    def 	__snake_case  (							self							:							Union[str, Any]      ):



       '''simple docstring'''







       snake_case       :				Optional[Any]		    =(
           self.local_path_to_dummy_data if self.use_local_dummy_data is True else self.github_path_to_dummy_data
       )

       snake_case       :				Dict		    =cached_path(
           _snake_case,							cache_dir=self.cache_dir,							extract_compressed_file=_snake_case,							force_extract=_snake_case      )

       return os.path.join(_snake_case,							self.dummy_file_name      )





    @property
    def 	__snake_case  (							self							:							Union[str, Any]      ):



       '''simple docstring'''







       return os.path.join(self.datasets_scripts_dir,							self.dataset_name,							self.dummy_zip_file      )





    @property
    def 	__snake_case  (							self							:							Union[str, Any]      ):



       '''simple docstring'''







       if self._bucket_url is None:
          snake_case       :				Optional[Any]		    =hf_github_url(self.dataset_name,							self.dummy_zip_file.replace(os.sep,							'''/'''      )      )
       return self._bucket_url





    @property
    def 	__snake_case  (							self							:							List[Any]      ):



       '''simple docstring'''







       if os.path.isdir(self.dummy_file      ):
          return self.dummy_file
       # else cut off path to file -> example `xsum`.
       return "/".join(self.dummy_file.replace(os.sep,							'''/'''      ).split('''/'''      )[:-1]      )





    def 	__snake_case  (							self							:							Tuple,							_snake_case							:							Tuple,							*_snake_case							:							Optional[Any]      ):



       '''simple docstring'''







       if self.load_existing_dummy_data:
          # dummy data is downloaded and tested
          snake_case       :				Any		    =self.dummy_file
       else:
          # dummy data cannot be downloaded and only the path to dummy file is returned
          snake_case       :				Optional[Any]		    =self.dummy_file_name

       # special case when data_url is a dict
       if isinstance(_snake_case,							_snake_case      ):
          return self.create_dummy_data_dict(_snake_case,							_snake_case      )
       elif isinstance(_snake_case,							(list, tuple)      ):
          return self.create_dummy_data_list(_snake_case,							_snake_case      )
       else:
          return self.create_dummy_data_single(_snake_case,							_snake_case      )





    def 	__snake_case  (							self							:							Any,							_snake_case							:							Any,							*_snake_case							:							int      ):



       '''simple docstring'''







       return self.download_and_extract(_snake_case      )





    def 	__snake_case  (							self							:							int,							_snake_case							:							int,							_snake_case							:							List[str]      ):



       '''simple docstring'''







       return self.download_and_extract(_snake_case      )





    def 	__snake_case  (							self							:							List[str],							_snake_case							:							List[str],							*_snake_case							:							Tuple,							**_snake_case							:							Tuple      ):



       '''simple docstring'''







       return path





    def 	__snake_case  (							self							:							Union[str, Any]      ):



       '''simple docstring'''







       return {}





    def 	__snake_case  (							self							:							str,							_snake_case							:							Optional[Any],							_snake_case							:							Tuple      ):



       '''simple docstring'''







       snake_case       :				int		    ={}
       for key, single_urls in data_url.items():
          for download_callback in self.download_callbacks:
             if isinstance(_snake_case,							_snake_case      ):
                for single_url in single_urls:
                   download_callback(_snake_case      )
             else:
                snake_case       :				List[Any]		    =single_urls
                download_callback(_snake_case      )
            # we force the name of each key to be the last file / folder name of the url path
            # if the url has arguments, we need to encode them with urllib.parse.quote_plus
          if isinstance(_snake_case,							_snake_case      ):
             snake_case       :				Optional[int]		    =[os.path.join(_snake_case,							urllib.parse.quote_plus(Path(_snake_case      ).name      )      ) for x in single_urls]
          else:
             snake_case       :				List[Any]		    =single_urls
             snake_case       :				List[str]		    =os.path.join(_snake_case,							urllib.parse.quote_plus(Path(_snake_case      ).name      )      )
          snake_case       :				List[Any]		    =value

       # make sure that values are unique
       if all(isinstance(_snake_case,							_snake_case      ) for i in dummy_data_dict.values()      ) and len(set(dummy_data_dict.values()      )      ) < len(
           dummy_data_dict.values()      ):
          # append key to value to make its name unique
          snake_case       :				Any		    ={key: value + key for key, value in dummy_data_dict.items()}

       return dummy_data_dict





    def 	__snake_case  (							self							:							Optional[int],							_snake_case							:							List[str],							_snake_case							:							Optional[Any]      ):



       '''simple docstring'''







       snake_case       :				Optional[Any]		    =[]
       # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
       snake_case       :				str		    =all(bool(re.findall('''[0-9]{3,}-of-[0-9]{3,}''',							_snake_case      )      ) for url in data_url      )
       snake_case       :				Union[str, Any]		    =all(
           url.startswith('''https://ftp.ncbi.nlm.nih.gov/pubmed/baseline/pubmed'''      ) for url in data_url      )
       if data_url and (is_tf_records or is_pubmed_records):
          snake_case       :				Any		    =[data_url[0]] * len(_snake_case      )
       for single_url in data_url:
          for download_callback in self.download_callbacks:
             download_callback(_snake_case      )
          # we force the name of each key to be the last file / folder name of the url path
          # if the url has arguments, we need to encode them with urllib.parse.quote_plus
          snake_case       :				List[Any]		    =os.path.join(_snake_case,							urllib.parse.quote_plus(single_url.split('''/'''      )[-1]      )      )
          dummy_data_list.append(_snake_case      )
       return dummy_data_list





    def 	__snake_case  (							self							:							List[str],							_snake_case							:							List[str],							_snake_case							:							Dict      ):



       '''simple docstring'''







       for download_callback in self.download_callbacks:
          download_callback(_snake_case      )
       # we force the name of each key to be the last file / folder name of the url path
       # if the url has arguments, we need to encode them with urllib.parse.quote_plus
       snake_case       :				Optional[Any]		    =os.path.join(_snake_case,							urllib.parse.quote_plus(data_url.split('''/'''      )[-1]      )      )
       if os.path.exists(_snake_case      ) or not self.load_existing_dummy_data:
          return value
       else:
          # Backward compatibility, maybe deprecate at one point.
          # For many datasets with single url calls to dl_manager.download_and_extract,
          # the dummy_data.zip file is actually the zipped downloaded file
          # while now we expected the dummy_data.zip file to be a directory containing
          # the downloaded file.
          return path_to_dummy_data





    def 	__snake_case  (							self							:							Any      ):



       '''simple docstring'''







       pass





    def 	__snake_case  (							self							:							Tuple      ):



       '''simple docstring'''







       pass





    def 	__snake_case  (							self							:							Optional[int],							_snake_case							:							Any      ):



       '''simple docstring'''







       def _iter_archive_members(_snake_case							:							int      ):
          # this preserves the order of the members inside the ZIP archive
          snake_case       :				List[str]		    =Path(self.dummy_file      ).parent
          snake_case       :				Union[str, Any]		    =path.relative_to(_snake_case      )
          with ZipFile(self.local_path_to_dummy_data      ) as zip_file:
             snake_case       :				str		    =zip_file.namelist()
          for member in members:
             if member.startswith(relative_path.as_posix()      ):
                yield dummy_parent_path.joinpath(_snake_case      )

       snake_case       :				List[str]		    =Path(_snake_case      )
       snake_case       :				List[str]		    =_iter_archive_members(_snake_case      ) if self.use_local_dummy_data else path.rglob('''*'''      )
       for file_path in file_paths:
          if file_path.is_file() and not file_path.name.startswith(('''.''', '''__''')      ):
             yield file_path.relative_to(_snake_case      ).as_posix(), file_path.open('''rb'''      )






    def 	__snake_case  (							self							:							int,							_snake_case							:							Tuple      ):



       '''simple docstring'''







       if not isinstance(_snake_case,							_snake_case      ):
          snake_case       :				int		    =[paths]
       for path in paths:
          if os.path.isfile(_snake_case      ):
             if os.path.basename(_snake_case      ).startswith(('''.''', '''__''')      ):
                return
             yield path
          else:
             for dirpath, dirnames, filenames in os.walk(_snake_case      ):
                if os.path.basename(_snake_case      ).startswith(('''.''', '''__''')      ):
                   continue
                dirnames.sort()
                for filename in sorted(_snake_case      ):
                   if filename.startswith(('''.''', '''__''')      ):
                      continue
                   yield os.path.join(_snake_case,							_snake_case      )









'''simple docstring'''






import warnings

from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding






class    __magic_name__				(						lowerCAmelCase							):
       UpperCAmelCase		 =["image_processor", "tokenizer"]
       UpperCAmelCase		 ="CLIPImageProcessor"
       UpperCAmelCase		 =("CLIPTokenizer", "CLIPTokenizerFast")






       def __init__(     self    ,				snake_case=None    ,				snake_case=None    ,				**snake_case)		->    Dict:




          '''simple docstring'''
          _UpperCAmelCase  :  Union[str, Any]				       =None
          if "feature_extractor" in kwargs:
             warnings.warn(
                 'The `feature_extractor` argument is deprecated and will be removed in v5, use `image_processor`'
                 ' instead.'    ,				snake_case    ,				)
             _UpperCAmelCase  :  str				       =kwargs.pop('feature_extractor')

          _UpperCAmelCase  :  List[str]				       =image_processor if image_processor is not None else feature_extractor
          if image_processor is None:
             raise ValueError('You need to specify an `image_processor`.')
          if tokenizer is None:
             raise ValueError('You need to specify a `tokenizer`.')

          super().__init__(snake_case    ,				snake_case)






       def __call__(     self    ,				snake_case=None    ,				snake_case=None    ,				snake_case=None    ,				**snake_case)		->    Optional[Any]:




          '''simple docstring'''

          if text is None and images is None:
             raise ValueError('You have to specify either text or images. Both cannot be none.')

          if text is not None:
             _UpperCAmelCase  :  List[Any]				       =self.tokenizer(snake_case    ,				return_tensors=snake_case    ,				**snake_case)

          if images is not None:
             _UpperCAmelCase  :  List[str]				       =self.image_processor(snake_case    ,				return_tensors=snake_case    ,				**snake_case)

          if text is not None and images is not None:
             _UpperCAmelCase  :  Any				       =image_features.pixel_values
             return encoding
          elif text is not None:
             return encoding
          else:
             return BatchEncoding(data=dict(**snake_case)    ,				tensor_type=snake_case)






       def      lowerCAmelCase						(     self    ,				*snake_case    ,				**snake_case)		->    int:




          '''simple docstring'''
          return self.tokenizer.batch_decode(*snake_case    ,				**snake_case)






       def      lowerCAmelCase						(     self    ,				*snake_case    ,				**snake_case)		->    Optional[int]:




          '''simple docstring'''
          return self.tokenizer.decode(*snake_case    ,				**snake_case)






       @property
       def      lowerCAmelCase						(     self)		->    List[Any]:




          '''simple docstring'''
          _UpperCAmelCase  :  List[Any]				       =self.tokenizer.model_input_names
          _UpperCAmelCase  :  Any				       =self.image_processor.model_input_names
          return list(dict.fromkeys(tokenizer_input_names + image_processor_input_names))






       @property
       def      lowerCAmelCase						(     self)		->    Any:




          '''simple docstring'''
          warnings.warn(
              '`feature_extractor_class` is deprecated and will be removed in v5. Use `image_processor_class` instead.'    ,				snake_case    ,				)
          return self.image_processor_class






       @property
       def      lowerCAmelCase						(     self)		->    Optional[int]:




          '''simple docstring'''
          warnings.warn(
              '`feature_extractor` is deprecated and will be removed in v5. Use `image_processor` instead.'    ,				snake_case    ,				)
          return self.image_processor

