




from typing import Dict

from transformers import EvalPrediction, HfArgumentParser, TrainingArguments, is_torch_available
from transformers.testing_utils import (
    TestCasePlus,
    execute_subprocess_async,
    get_torch_dist_unique_port,
    require_torch_multi_gpu,
    require_torch_neuroncore,
)
from transformers.training_args import ParallelMode
from transformers.utils import logging


_snake_case            = logging.get_logger(__name__)


if is_torch_available():
  import torch
  from torch import nn
  from torch.utils.data import Dataset

  from transformers import Trainer







  class 		lowercase							(     UpperCamelCase__	):

       def __init__(						self	,  _a = 101       )		->							str:
             _A							:  int    =		length

       def __len__(						self       )		->							List[Any]:
             return self.length



       def __getitem__(						self	,  _a       )		->							List[str]:
             return i







  class 		lowercase							:

       def __call__(						self	,  _a       )		->							int:
             return {"input_ids": torch.tensor(_a       ), "labels": torch.tensor(_a       )}







  class 		lowercase							(     nn.Module	):

       def __init__(						self       )		->							Optional[Any]:
             super().__init__()
             # Add some (unused) params otherwise DDP will complain.
             _A							:  Union[str, Any]    =		nn.Linear(120	,  80       )



       def       a__			(						self	,  _a	,  _a=None       )		->							str:







             if labels is not None:
                   return torch.tensor(0.0	,  device=input_ids.device       ), input_ids
             else:
                   return input_ids







class 		lowercase							(     UpperCamelCase__	):

     @require_torch_neuroncore
     def       a__			(						self       )		->							int:
           _A							:  Union[str, Any]    =		F'''--nproc_per_node=2
            --master_port={get_torch_dist_unique_port()}
            {self.test_file_dir}/test_trainer_distributed.py
        '''.split()
           _A							:  str    =		self.get_auto_remove_tmp_dir()
           _A							:  str    =		F'''--output_dir {output_dir}'''.split()
           _A							:  Optional[Any]    =		['torchrun'] + distributed_args + args
           execute_subprocess_async(_a	,  env=self.get_env()       )
           # successful return here == success - any errors would have caused an error in the sub-call







class 		lowercase							(     UpperCamelCase__	):

     @require_torch_multi_gpu
     def       a__			(						self       )		->							Optional[Any]:
           _A							:  int    =		F'''--nproc_per_node={torch.cuda.device_count()}
            --master_port={get_torch_dist_unique_port()}
            {self.test_file_dir}/test_trainer_distributed.py
        '''.split()
           _A							:  Union[str, Any]    =		self.get_auto_remove_tmp_dir()
           _A							:  List[Any]    =		F'''--output_dir {output_dir}'''.split()
           _A							:  Optional[int]    =		['torchrun'] + distributed_args + args
           execute_subprocess_async(_a	,  env=self.get_env()       )
           # successful return here == success - any errors would have caused an error in the sub-call


if __name__ == "__main__":
  # The script below is meant to be run under torch.distributed, on a machine with multiple GPUs:
  #
  # PYTHONPATH="src" python -m torch.distributed.run --nproc_per_node 2 --output_dir output_dir ./tests/test_trainer_distributed.py

  _snake_case            = HfArgumentParser((TrainingArguments,))
  _snake_case            = parser.parse_args_into_dataclasses()[0]

  logger.warning(
      f"""Process rank: {training_args.local_rank}, device: {training_args.device}, n_gpu: {training_args.n_gpu}, """
      f"""distributed training: {training_args.parallel_mode != ParallelMode.NOT_DISTRIBUTED}"""
  )

  # Essentially, what we want to verify in the distributed case is that we get all samples back,
  # in the right order. (this is crucial for prediction for instance)
  for dataset_length in [101, 40, 7]:
    _snake_case            = DummyDataset(dataset_length)







    def        lowerCAmelCase_			(  snake_case_  ):
          _A							:  Union[str, Any]    =		list(range(len(SCREAMING_SNAKE_CASE_  )  )  )
          _A							:  Dict    =		p.predictions.tolist() == sequential and p.label_ids.tolist() == sequential
          if not success and training_args.local_rank == 0:
                logger.warning(
                    """Predictions and/or labels do not match expected results:\n  - predictions: """
                    f'''{p.predictions.tolist()}\n  - labels: {p.label_ids.tolist()}\n  - expected: {sequential}'''  )
          return {"success": success}

    _snake_case            = Trainer(
        model=DummyModel(),
        args=training_args,
        data_collator=DummyDataCollator(),
        eval_dataset=dataset,
        compute_metrics=compute_metrics,
    )
    _snake_case            = trainer.evaluate()
    logger.info(metrics)
    if metrics["eval_success"] is not True:
      logger.error(metrics)
      exit(1)

    _snake_case            = trainer.predict(dataset)
    logger.info(p.metrics)
    if p.metrics["test_success"] is not True:
      logger.error(p.metrics)
      exit(1)

    _snake_case            = 2

    _snake_case            = trainer.evaluate()
    logger.info(metrics)
    if metrics["eval_success"] is not True:
      logger.error(metrics)
      exit(1)

    _snake_case            = trainer.predict(dataset)
    logger.info(p.metrics)
    if p.metrics["test_success"] is not True:
      logger.error(p.metrics)
      exit(1)







    _snake_case            = None







import json
import os
import unittest

from transformers import CLIPTokenizer, CLIPTokenizerFast
from transformers.models.clip.tokenization_clip import VOCAB_FILES_NAMES
from transformers.testing_utils import require_ftfy, require_tokenizers

from ...test_tokenization_common import TokenizerTesterMixin







@require_tokenizers
class 		lowercase							(     UpperCamelCase__,unittest.TestCase	):
     _a											=		CLIPTokenizer
     _a											=		CLIPTokenizerFast
     _a											=		True
     _a											=		{}
     _a											=		False

     def       a__			(						self       )		->							Optional[Any]:
           super().setUp()

           # fmt: off
           _A							:  int    =		["""l""", """o""", """w""", """e""", """r""", """s""", """t""", """i""", """d""", """n""", """lo""", """l</w>""", """w</w>""", """r</w>""", """t</w>""", """low</w>""", """er</w>""", """lowest</w>""", """newer</w>""", """wider""", """<unk>""", """<|startoftext|>""", """<|endoftext|>"""]
           # fmt: on
           _A							:  str    =		dict(zip(_a	,  range(len(_a       )       )       )       )
           _A							:  Optional[int]    =		["""#version: 0.2""", """l o""", """lo w</w>""", """e r</w>"""]
           _A							:  str    =		{"""unk_token""": """<unk>"""}

           _A							:  Union[str, Any]    =		os.path.join(self.tmpdirname	,  VOCAB_FILES_NAMES["""vocab_file"""]       )
           _A							:  Union[str, Any]    =		os.path.join(self.tmpdirname	,  VOCAB_FILES_NAMES["""merges_file"""]       )
           with open(self.vocab_file	,  """w"""	,  encoding="""utf-8"""       ) as fp:
                 fp.write(json.dumps(_a       ) + """\n"""       )
           with open(self.merges_file	,  """w"""	,  encoding="""utf-8"""       ) as fp:
                 fp.write("""\n""".join(_a       )       )

     def       a__			(						self	,  **_a       )		->							List[Any]:
           kwargs.update(self.special_tokens_map       )
           return CLIPTokenizer.from_pretrained(self.tmpdirname	,  **_a       )

     def       a__			(						self	,  **_a       )		->							Union[str, Any]:
           kwargs.update(self.special_tokens_map       )
           return CLIPTokenizerFast.from_pretrained(self.tmpdirname	,  **_a       )

     def       a__			(						self	,  _a       )		->							str:
           _A							:  Tuple    =		"""lower newer"""
           _A							:  Optional[Any]    =		"""lower newer"""
           return input_text, output_text

     def       a__			(						self       )		->							List[Any]:
           _A							:  Optional[int]    =		CLIPTokenizer(self.vocab_file	,  self.merges_file	,  **self.special_tokens_map       )
           _A							:  str    =		"""lower newer"""
           _A							:  List[str]    =		["""lo""", """w""", """er</w>""", """n""", """e""", """w""", """er</w>"""]
           _A							:  int    =		tokenizer.tokenize(_a       )
           self.assertListEqual(_a	,  _a       )

           _A							:  str    =		tokens + [tokenizer.unk_token]
           _A							:  Dict    =		[10, 2, 16, 9, 3, 2, 16, 20]
           self.assertListEqual(tokenizer.convert_tokens_to_ids(_a       )	,  _a       )

     @require_ftfy
     def       a__			(						self       )		->							Tuple:
           for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
                 with self.subTest(F'''{tokenizer.__class__.__name__} ({pretrained_name})'''       ):
                       _A							:  Dict    =		self.tokenizer_class.from_pretrained(_a	,  **_a       )
                       _A							:  str    =		self.rust_tokenizer_class.from_pretrained(_a	,  **_a       )

                       _A							:  List[str]    =		"""A\n'll 11p223RF☆ho!!to?'d'd''d of a cat to-$''d."""
                       _A							:  Dict    =		tokenizer_s.tokenize(_a       )
                       _A							:  Dict    =		tokenizer_r.tokenize(_a       )

                       self.assertListEqual(_a	,  _a       )

                       # Test that the tokenization is identical on an example containing a character (Latin Small Letter A
                       # with Tilde) encoded in 2 different ways
                       _A							:  Tuple    =		"""xa\u0303y""" + """ """ + """x\xe3y"""
                       _A							:  Dict    =		tokenizer_s.tokenize(_a       )
                       _A							:  Dict    =		tokenizer_r.tokenize(_a       )

                       self.assertListEqual(_a	,  _a       )

                       # Test that the tokenization is identical on unicode of space type
                       _A							:  Any    =		[
                           """\u0009""",  # (horizontal tab, '\t')
                           """\u000B""",  # (vertical tab)
                           """\u000C""",  # (form feed)
                           """\u0020""",  # (space, ' ')
                           """\u200E""",  # (left-to-right mark):w
                           """\u200F""",  # (right-to-left mark)
                       ]
                       for unicode_seq in spaces_unicodes:
                             _A							:  Optional[int]    =		tokenizer_s.tokenize(_a       )
                             _A							:  List[Any]    =		tokenizer_r.tokenize(_a       )

                             self.assertListEqual(_a	,  _a       )

                       # Test that the tokenization is identical on unicode of line break type
                       _A							:  int    =		[
                           """\u000A""",  # (line feed, '\n')
                           """\r\n""",  # (carriage return and line feed, '\r\n')
                           """\u000D""",  # (carriage return, '\r')
                           """\r""",  # (carriage return, '\r')
                           """\u000D""",  # (carriage return, '\r')
                           """\u2028""",  # (line separator)
                           """\u2029""",  # (paragraph separator)
                           # "\u0085", # (next line)
                       ]

                       # The tokenization is not identical for the character "\u0085" (next line). The slow version using ftfy transforms
                       # it into the Horizontal Ellipsis character "…" ("\u2026") while the fast version transforms it into a
                       # space (and thus into an empty list).

                       for unicode_seq in line_break_unicodes:
                             _A							:  Optional[Any]    =		tokenizer_s.tokenize(_a       )
                             _A							:  Tuple    =		tokenizer_r.tokenize(_a       )

                             self.assertListEqual(_a	,  _a       )

     def       a__			(						self       )		->							Any:
           # Test which aims to verify that the offsets are well adapted to the argument `add_prefix_space`
           for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
                 with self.subTest(F'''{tokenizer.__class__.__name__} ({pretrained_name})'''       ):
                       _A							:  List[Any]    =		"""hello"""  # `hello` is a token in the vocabulary of `pretrained_name`
                       _A							:  str    =		F'''{text_of_1_token} {text_of_1_token}'''

                       _A							:  str    =		self.rust_tokenizer_class.from_pretrained(
                           _a	,  use_fast=_a	,  )
                       _A							:  Dict    =		tokenizer_r(_a	,  return_offsets_mapping=_a	,  add_special_tokens=_a       )
                       self.assertEqual(encoding.offset_mapping[0]	,  (0, len(_a       ))       )
                       self.assertEqual(
                           encoding.offset_mapping[1]	,  (len(_a       ) + 1, len(_a       ) + 1 + len(_a       ))	,  )

                       _A							:  Any    =		F''' {text}'''

                       _A							:  List[Any]    =		self.rust_tokenizer_class.from_pretrained(
                           _a	,  use_fast=_a	,  )
                       _A							:  Optional[int]    =		tokenizer_r(_a	,  return_offsets_mapping=_a	,  add_special_tokens=_a       )
                       self.assertEqual(encoding.offset_mapping[0]	,  (1, 1 + len(_a       ))       )
                       self.assertEqual(
                           encoding.offset_mapping[1]	,  (1 + len(_a       ) + 1, 1 + len(_a       ) + 1 + len(_a       ))	,  )

     def       a__			(						self       )		->							List[Any]:
           # Test related to the breaking change introduced in transformers v4.17.0
           # We need to check that an error in raised when the user try to load a previous version of the tokenizer.
           with self.assertRaises(_a       ) as context:
                 self.rust_tokenizer_class.from_pretrained("""robot-test/old-clip-tokenizer"""       )

           self.assertTrue(
               context.exception.args[0].startswith(
                   """The `backend_tokenizer` provided does not match the expected format."""       )       )

     @require_ftfy
     def       a__			(						self       )		->							str:
           super().test_tokenization_python_rust_equals()



     def       a__			(						self       )		->							Union[str, Any]:
           # CLIP always lower cases letters
           pass

