





'''simple docstring'''






# Lint as: python3
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 					__snake_case  :






   '''simple docstring'''





   lowerCamelCase__    							=      '''dummy_data'''
   lowerCamelCase__    							=      '''datasets'''
   lowerCamelCase__    							=      False

   def __init__(							self	,		__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE = None	,		__SCREAMING_SNAKE_CASE = False	,		__SCREAMING_SNAKE_CASE = True	,		__SCREAMING_SNAKE_CASE = None	,		):
    snake_case__     :					List[Any]    =  0
    snake_case__     :					Union[str, Any]    =  dataset_name
    snake_case__     :					Optional[int]    =  cache_dir
    snake_case__     :					Union[str, Any]    =  use_local_dummy_data
    snake_case__     :					int    =  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__     :					Union[str, Any]    =  load_existing_dummy_data

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

   @property
   def 		__UpperCamelCase						(							self       ):
    if self._dummy_file is None:
     snake_case__     :					List[str]    =  self.download_dummy_data()
    return self._dummy_file

   @property
   def 		__UpperCamelCase						(							self       ):
    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 		__UpperCamelCase						(							self       ):
    return os.path.join(self.dummy_data_folder	,		"""dummy_data.zip"""       )

   def 		__UpperCamelCase						(							self       ):
    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__     :					Optional[int]    =  cached_path(
        __SCREAMING_SNAKE_CASE	,		cache_dir=self.cache_dir	,		extract_compressed_file=__SCREAMING_SNAKE_CASE	,		force_extract=__SCREAMING_SNAKE_CASE       )

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

   @property
   def 		__UpperCamelCase						(							self       ):
    return os.path.join(self.datasets_scripts_dir	,		self.dataset_name	,		self.dummy_zip_file       )

   @property
   def 		__UpperCamelCase						(							self       ):
    if self._bucket_url is None:
     snake_case__     :					List[str]    =  hf_github_url(self.dataset_name	,		self.dummy_zip_file.replace(os.sep	,		"""/"""       )       )
    return self._bucket_url

   @property
   def 		__UpperCamelCase						(							self       ):
    # return full path if its a dir
    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 		__UpperCamelCase						(							self	,		__SCREAMING_SNAKE_CASE	,		*__SCREAMING_SNAKE_CASE       ):
    if self.load_existing_dummy_data:
     # dummy data is downloaded and tested
     snake_case__     :					List[Any]    =  self.dummy_file
    else:
     # dummy data cannot be downloaded and only the path to dummy file is returned
     snake_case__     :					List[Any]    =  self.dummy_file_name

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

   def 		__UpperCamelCase						(							self	,		__SCREAMING_SNAKE_CASE	,		*__SCREAMING_SNAKE_CASE       ):
    return self.download_and_extract(__SCREAMING_SNAKE_CASE       )

   def 		__UpperCamelCase						(							self	,		__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE       ):
    return self.download_and_extract(__SCREAMING_SNAKE_CASE       )

   def 		__UpperCamelCase						(							self	,		__SCREAMING_SNAKE_CASE	,		*__SCREAMING_SNAKE_CASE	,		**__SCREAMING_SNAKE_CASE       ):
    return path

   def 		__UpperCamelCase						(							self       ):
    return {}

   def 		__UpperCamelCase						(							self	,		__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE       ):
    snake_case__     :					int    =  {}
    for key, single_urls in data_url.items():
     for download_callback in self.download_callbacks:
      if isinstance(__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE       ):
       for single_url in single_urls:
        download_callback(__SCREAMING_SNAKE_CASE       )
      else:
       snake_case__     :					List[str]    =  single_urls
       download_callback(__SCREAMING_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(__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE       ):
      snake_case__     :					Tuple    =  [os.path.join(__SCREAMING_SNAKE_CASE	,		urllib.parse.quote_plus(Path(__SCREAMING_SNAKE_CASE       ).name       )       ) for x in single_urls]
     else:
      snake_case__     :					List[Any]    =  single_urls
      snake_case__     :					Tuple    =  os.path.join(__SCREAMING_SNAKE_CASE	,		urllib.parse.quote_plus(Path(__SCREAMING_SNAKE_CASE       ).name       )       )
     snake_case__     :					Optional[int]    =  value

    # make sure that values are unique
    if all(isinstance(__SCREAMING_SNAKE_CASE	,		__SCREAMING_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__     :					List[Any]    =  {key: value + key for key, value in dummy_data_dict.items()}

    return dummy_data_dict

   def 		__UpperCamelCase						(							self	,		__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE       ):
    snake_case__     :					Dict    =  []
    # trick: if there are many shards named like `data.txt-000001-of-00300`, only use the first one
    snake_case__     :					Tuple    =  all(bool(re.findall("""[0-9]{3,}-of-[0-9]{3,}"""	,		__SCREAMING_SNAKE_CASE       )       ) for url in data_url       )
    snake_case__     :					List[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__     :					List[str]    =  [data_url[0]] * len(__SCREAMING_SNAKE_CASE       )
    for single_url in data_url:
     for download_callback in self.download_callbacks:
      download_callback(__SCREAMING_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(__SCREAMING_SNAKE_CASE	,		urllib.parse.quote_plus(single_url.split("""/"""       )[-1]       )       )
     dummy_data_list.append(__SCREAMING_SNAKE_CASE       )
    return dummy_data_list

   def 		__UpperCamelCase						(							self	,		__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE       ):
    for download_callback in self.download_callbacks:
     download_callback(__SCREAMING_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__     :					Any    =  os.path.join(__SCREAMING_SNAKE_CASE	,		urllib.parse.quote_plus(data_url.split("""/"""       )[-1]       )       )
    if os.path.exists(__SCREAMING_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 		__UpperCamelCase						(							self       ):
    pass

   def 		__UpperCamelCase						(							self       ):
    pass

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

    snake_case__     :					Any    =  Path(__SCREAMING_SNAKE_CASE       )
    snake_case__     :					int    =  _iter_archive_members(__SCREAMING_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(__SCREAMING_SNAKE_CASE       ).as_posix(), file_path.open("""rb"""       )


   def 		__UpperCamelCase						(							self	,		__SCREAMING_SNAKE_CASE       ):
    if not isinstance(__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE       ):
     snake_case__     :					Optional[int]    =  [paths]
    for path in paths:
     if os.path.isfile(__SCREAMING_SNAKE_CASE       ):
      if os.path.basename(__SCREAMING_SNAKE_CASE       ).startswith((""".""", """__""")       ):
       return
      yield path
     else:
      for dirpath, dirnames, filenames in os.walk(__SCREAMING_SNAKE_CASE       ):
       if os.path.basename(__SCREAMING_SNAKE_CASE       ).startswith((""".""", """__""")       ):
        continue
       dirnames.sort()
       for filename in sorted(__SCREAMING_SNAKE_CASE       ):
        if filename.startswith((""".""", """__""")       ):
         continue
        yield os.path.join(__SCREAMING_SNAKE_CASE	,		__SCREAMING_SNAKE_CASE       )











import json
import os
import shutil
import tempfile
import unittest

import numpy as np

from transformers import BertTokenizerFast
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES, BertTokenizer
from transformers.testing_utils import require_tokenizers, require_vision
from transformers.utils import IMAGE_PROCESSOR_NAME, is_vision_available


if is_vision_available():
    from PIL import Image

    from transformers import VisionTextDualEncoderProcessor, ViTImageProcessor

@require_tokenizers
@require_vision
class   lowerCAmelCase				(   unittest.TestCase     ):
 def 	A_      ( self    : str      )     ->      Dict:
     lowerCamelCase__       : Optional[Any]   					=						tempfile.mkdtemp()

     # fmt: off
     lowerCamelCase__       : Tuple   					=						['[UNK]', '[CLS]', '[SEP]', '[PAD]', '[MASK]', 'want', '##want', '##ed', 'wa', 'un', 'runn', '##ing', ',', 'low', 'lowest']
     # fmt: on
     lowerCamelCase__       : int   					=						os.path.join(self.tmpdirname     ,	VOCAB_FILES_NAMES['vocab_file']      )
     with open(self.vocab_file     ,	'w'     ,	encoding='utf-8'      ) as vocab_writer:
         vocab_writer.write(''.join([x + '\n' for x in vocab_tokens]      )      )

     lowerCamelCase__       : Any   					=						{
         'do_resize': True,
         'size': {'height': 18, 'width': 18},
         'do_normalize': True,
         'image_mean': [0.5, 0.5, 0.5],
         'image_std': [0.5, 0.5, 0.5],
     }
     lowerCamelCase__       : int   					=						os.path.join(self.tmpdirname     ,	UpperCAmelCase      )
     with open(self.image_processor_file     ,	'w'     ,	encoding='utf-8'      ) as fp:
         json.dump(UpperCAmelCase     ,	UpperCAmelCase      )
 def 	A_      ( self    : List[str]     ,	**UpperCAmelCase    : str      )     ->      int:
     return BertTokenizer.from_pretrained(self.tmpdirname     ,	**UpperCAmelCase      )
 def 	A_      ( self    : Optional[int]     ,	**UpperCAmelCase    : List[str]      )     ->      List[str]:
     return ViTImageProcessor.from_pretrained(self.tmpdirname     ,	**UpperCAmelCase      )
 def 	A_      ( self    : int      )     ->      Any:
     shutil.rmtree(self.tmpdirname      )
 def 	A_      ( self    : str      )     ->      str:
     lowerCamelCase__       : List[Any]   					=						[np.random.randint(255     ,	size=(3, 30, 400)     ,	dtype=np.uinta      )]

     lowerCamelCase__       : Union[str, Any]   					=						[Image.fromarray(np.moveaxis(UpperCAmelCase     ,	0     ,	-1      )      ) for x in image_inputs]

     return image_inputs
 def 	A_      ( self    : Dict      )     ->      Dict:
     lowerCamelCase__       : List[Any]   					=						self.get_tokenizer()
     lowerCamelCase__       : Tuple   					=						self.get_image_processor()

     lowerCamelCase__       : Union[str, Any]   					=						VisionTextDualEncoderProcessor(tokenizer=UpperCAmelCase     ,	image_processor=UpperCAmelCase      )

     processor.save_pretrained(self.tmpdirname      )
     lowerCamelCase__       : List[str]   					=						VisionTextDualEncoderProcessor.from_pretrained(self.tmpdirname      )

     self.assertEqual(processor.tokenizer.get_vocab()     ,	tokenizer.get_vocab()      )
     self.assertIsInstance(processor.tokenizer     ,	(BertTokenizer, BertTokenizerFast)      )

     self.assertEqual(processor.image_processor.to_json_string()     ,	image_processor.to_json_string()      )
     self.assertIsInstance(processor.image_processor     ,	UpperCAmelCase      )
 def 	A_      ( self    : List[str]      )     ->      Tuple:
     lowerCamelCase__       : List[Any]   					=						VisionTextDualEncoderProcessor(
         tokenizer=self.get_tokenizer()     ,	image_processor=self.get_image_processor()      )
     processor.save_pretrained(self.tmpdirname      )

     lowerCamelCase__       : Tuple   					=						self.get_tokenizer(bos_token='(BOS)'     ,	eos_token='(EOS)'      )
     lowerCamelCase__       : int   					=						self.get_image_processor(do_normalize=UpperCAmelCase     ,	padding_value=1.0      )

     lowerCamelCase__       : Union[str, Any]   					=						VisionTextDualEncoderProcessor.from_pretrained(
         self.tmpdirname     ,	bos_token='(BOS)'     ,	eos_token='(EOS)'     ,	do_normalize=UpperCAmelCase     ,	padding_value=1.0      )

     self.assertEqual(processor.tokenizer.get_vocab()     ,	tokenizer_add_kwargs.get_vocab()      )
     self.assertIsInstance(processor.tokenizer     ,	(BertTokenizer, BertTokenizerFast)      )

     self.assertEqual(processor.image_processor.to_json_string()     ,	image_processor_add_kwargs.to_json_string()      )
     self.assertIsInstance(processor.image_processor     ,	UpperCAmelCase      )
 def 	A_      ( self    : Any      )     ->      Tuple:
     lowerCamelCase__       : List[Any]   					=						self.get_image_processor()
     lowerCamelCase__       : Any   					=						self.get_tokenizer()

     lowerCamelCase__       : Optional[int]   					=						VisionTextDualEncoderProcessor(tokenizer=UpperCAmelCase     ,	image_processor=UpperCAmelCase      )

     lowerCamelCase__       : Optional[int]   					=						self.prepare_image_inputs()

     lowerCamelCase__       : int   					=						image_processor(UpperCAmelCase     ,	return_tensors='np'      )
     lowerCamelCase__       : List[str]   					=						processor(images=UpperCAmelCase     ,	return_tensors='np'      )

     for key in input_feat_extract.keys():
         self.assertAlmostEqual(input_feat_extract[key].sum()     ,	input_processor[key].sum()     ,	delta=1e-2      )
 def 	A_      ( self    : Tuple      )     ->      List[Any]:
     lowerCamelCase__       : Dict   					=						self.get_image_processor()
     lowerCamelCase__       : List[str]   					=						self.get_tokenizer()

     lowerCamelCase__       : str   					=						VisionTextDualEncoderProcessor(tokenizer=UpperCAmelCase     ,	image_processor=UpperCAmelCase      )

     lowerCamelCase__       : Tuple   					=						'lower newer'

     lowerCamelCase__       : Tuple   					=						processor(text=UpperCAmelCase      )

     lowerCamelCase__       : Any   					=						tokenizer(UpperCAmelCase      )

     for key in encoded_tok.keys():
         self.assertListEqual(encoded_tok[key]     ,	encoded_processor[key]      )
 def 	A_      ( self    : Optional[int]      )     ->      Any:
     lowerCamelCase__       : Optional[int]   					=						self.get_image_processor()
     lowerCamelCase__       : Tuple   					=						self.get_tokenizer()

     lowerCamelCase__       : Tuple   					=						VisionTextDualEncoderProcessor(tokenizer=UpperCAmelCase     ,	image_processor=UpperCAmelCase      )

     lowerCamelCase__       : str   					=						'lower newer'
     lowerCamelCase__       : Dict   					=						self.prepare_image_inputs()

     lowerCamelCase__       : int   					=						processor(text=UpperCAmelCase     ,	images=UpperCAmelCase      )

     self.assertListEqual(list(inputs.keys()      )     ,	['input_ids', 'token_type_ids', 'attention_mask', 'pixel_values']      )

     # test if it raises when no input is passed
     with self.assertRaises(UpperCAmelCase      ):
         processor()
 def 	A_      ( self    : Optional[Any]      )     ->      List[str]:
     lowerCamelCase__       : Dict   					=						self.get_image_processor()
     lowerCamelCase__       : Union[str, Any]   					=						self.get_tokenizer()

     lowerCamelCase__       : List[str]   					=						VisionTextDualEncoderProcessor(tokenizer=UpperCAmelCase     ,	image_processor=UpperCAmelCase      )

     lowerCamelCase__       : List[str]   					=						[[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

     lowerCamelCase__       : Dict   					=						processor.batch_decode(UpperCAmelCase      )
     lowerCamelCase__       : List[str]   					=						tokenizer.batch_decode(UpperCAmelCase      )

     self.assertListEqual(UpperCAmelCase     ,	UpperCAmelCase      )







 def 	A_      ( self    : Optional[int]      )     ->      List[str]:
     lowerCamelCase__       : Tuple   					=						self.get_image_processor()
     lowerCamelCase__       : List[str]   					=						self.get_tokenizer()

     lowerCamelCase__       : Union[str, Any]   					=						VisionTextDualEncoderProcessor(tokenizer=UpperCAmelCase     ,	image_processor=UpperCAmelCase      )

     lowerCamelCase__       : Dict   					=						'lower newer'
     lowerCamelCase__       : List[Any]   					=						self.prepare_image_inputs()

     lowerCamelCase__       : Union[str, Any]   					=						processor(text=UpperCAmelCase     ,	images=UpperCAmelCase      )

     self.assertListEqual(list(inputs.keys()      )     ,	processor.model_input_names      )





