

from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
      import tensorflow as tf

      from transformers import (
          TF_XLM_PRETRAINED_MODEL_ARCHIVE_LIST,
          TFXLMForMultipleChoice,
          TFXLMForQuestionAnsweringSimple,
          TFXLMForSequenceClassification,
          TFXLMForTokenClassification,
          TFXLMModel,
          TFXLMWithLMHeadModel,
          XLMConfig,
      )

class      _a				:




      """simple docstring"""

      def __init__(      self       :							Optional[Any]	, UpperCAmelCase       :							Union[str, Any]	, ):
            A_	     =							parent
            A_	     =							13
            A_	     =							7
            A_	     =							True
            A_	     =							True
            A_	     =							True
            A_	     =							True
            A_	     =							True
            A_	     =							False
            A_	     =							False
            A_	     =							False
            A_	     =							2
            A_	     =							99
            A_	     =							0
            A_	     =							32
            A_	     =							2
            A_	     =							4
            A_	     =							0.1
            A_	     =							0.1
            A_	     =							512
            A_	     =							16
            A_	     =							2
            A_	     =							0.02
            A_	     =							3
            A_	     =							4
            A_	     =							"last"
            A_	     =							True
            A_	     =							None
            A_	     =							0





      def 		__A  (      self       :							int  ):
            A_	     =							ids_tensor([self.batch_size, self.seq_length]	, self.vocab_size  )
            A_	     =							random_attention_mask([self.batch_size, self.seq_length]	, dtype=tf.floataa  )

            A_	     =							None
            if self.use_input_lengths:
                  A_	     =							(
                      ids_tensor([self.batch_size]	, vocab_size=2  ) + self.seq_length - 2
                  )  # small variation of seq_length

            A_	     =							None
            if self.use_token_type_ids:
                  A_	     =							ids_tensor([self.batch_size, self.seq_length]	, self.n_langs  )

            A_	     =							None
            A_	     =							None
            A_	     =							None
            if self.use_labels:
                  A_	     =							ids_tensor([self.batch_size]	, self.type_sequence_label_size  )
                  A_	     =							ids_tensor([self.batch_size, self.seq_length]	, self.num_labels  )
                  A_	     =							ids_tensor([self.batch_size]	, 2	, dtype=tf.floataa  )
                  A_	     =							ids_tensor([self.batch_size]	, self.num_choices  )

            A_	     =							XLMConfig(
                vocab_size=self.vocab_size	, n_special=self.n_special	, emb_dim=self.hidden_size	, n_layers=self.num_hidden_layers	, n_heads=self.num_attention_heads	, dropout=self.hidden_dropout_prob	, attention_dropout=self.attention_probs_dropout_prob	, gelu_activation=self.gelu_activation	, sinusoidal_embeddings=self.sinusoidal_embeddings	, asm=self.asm	, causal=self.causal	, n_langs=self.n_langs	, max_position_embeddings=self.max_position_embeddings	, initializer_range=self.initializer_range	, summary_type=self.summary_type	, use_proj=self.use_proj	, bos_token_id=self.bos_token_id	, )

            return (
                config,
                input_ids,
                token_type_ids,
                input_lengths,
                sequence_labels,
                token_labels,
                is_impossible_labels,
                choice_labels,
                input_mask,
            )





      def 		__A  (      self       :							List[str]	, UpperCAmelCase       :							Any	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							Dict	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							Any	, UpperCAmelCase       :							int	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							Any	, ):
            A_	     =							TFXLMModel(config=UpperCAmelCase  )
            A_	     =							{"input_ids": input_ids, "lengths": input_lengths, "langs": token_type_ids}
            A_	     =							model(UpperCAmelCase  )

            A_	     =							[input_ids, input_mask]
            A_	     =							model(UpperCAmelCase  )
            self.parent.assertEqual(result.last_hidden_state.shape	, (self.batch_size, self.seq_length, self.hidden_size)  )





      def 		__A  (      self       :							Dict	, UpperCAmelCase       :							List[str]	, UpperCAmelCase       :							List[Any]	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							int	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							List[Any]	, UpperCAmelCase       :							str	, ):
            A_	     =							TFXLMWithLMHeadModel(UpperCAmelCase  )

            A_	     =							{"input_ids": input_ids, "lengths": input_lengths, "langs": token_type_ids}
            A_	     =							model(UpperCAmelCase  )

            A_	     =							outputs

            self.parent.assertEqual(result.logits.shape	, (self.batch_size, self.seq_length, self.vocab_size)  )





      def 		__A  (      self       :							Tuple	, UpperCAmelCase       :							str	, UpperCAmelCase       :							List[str]	, UpperCAmelCase       :							List[str]	, UpperCAmelCase       :							Tuple	, UpperCAmelCase       :							List[str]	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							Dict	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							int	, ):
            A_	     =							TFXLMForQuestionAnsweringSimple(UpperCAmelCase  )

            A_	     =							{"input_ids": input_ids, "lengths": input_lengths}

            A_	     =							model(UpperCAmelCase  )

            self.parent.assertEqual(result.start_logits.shape	, (self.batch_size, self.seq_length)  )
            self.parent.assertEqual(result.end_logits.shape	, (self.batch_size, self.seq_length)  )





      def 		__A  (      self       :							Dict	, UpperCAmelCase       :							Tuple	, UpperCAmelCase       :							List[Any]	, UpperCAmelCase       :							Any	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							List[Any]	, UpperCAmelCase       :							Any	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							Union[str, Any]	, ):
            A_	     =							TFXLMForSequenceClassification(UpperCAmelCase  )

            A_	     =							{"input_ids": input_ids, "lengths": input_lengths}

            A_	     =							model(UpperCAmelCase  )

            self.parent.assertEqual(result.logits.shape	, (self.batch_size, self.type_sequence_label_size)  )





      def 		__A  (      self       :							Tuple	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							Optional[Any]	, UpperCAmelCase       :							Tuple	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							int	, UpperCAmelCase       :							int	, UpperCAmelCase       :							Tuple	, UpperCAmelCase       :							Dict	, UpperCAmelCase       :							List[str]	, ):
            A_	     =							self.num_labels
            A_	     =							TFXLMForTokenClassification(config=UpperCAmelCase  )
            A_	     =							{"input_ids": input_ids, "attention_mask": input_mask, "token_type_ids": token_type_ids}
            A_	     =							model(UpperCAmelCase  )
            self.parent.assertEqual(result.logits.shape	, (self.batch_size, self.seq_length, self.num_labels)  )





      def 		__A  (      self       :							List[Any]	, UpperCAmelCase       :							str	, UpperCAmelCase       :							List[str]	, UpperCAmelCase       :							Tuple	, UpperCAmelCase       :							str	, UpperCAmelCase       :							List[str]	, UpperCAmelCase       :							Tuple	, UpperCAmelCase       :							Any	, UpperCAmelCase       :							int	, UpperCAmelCase       :							List[str]	, ):
            A_	     =							self.num_choices
            A_	     =							TFXLMForMultipleChoice(config=UpperCAmelCase  )
            A_	     =							tf.tile(tf.expand_dims(UpperCAmelCase	, 1  )	, (1, self.num_choices, 1)  )
            A_	     =							tf.tile(tf.expand_dims(UpperCAmelCase	, 1  )	, (1, self.num_choices, 1)  )
            A_	     =							tf.tile(tf.expand_dims(UpperCAmelCase	, 1  )	, (1, self.num_choices, 1)  )
            A_	     =							{
                "input_ids": multiple_choice_inputs_ids,
                "attention_mask": multiple_choice_input_mask,
                "token_type_ids": multiple_choice_token_type_ids,
            }
            A_	     =							model(UpperCAmelCase  )
            self.parent.assertEqual(result.logits.shape	, (self.batch_size, self.num_choices)  )







      def 		__A  (      self       :							str  ):
            A_	     =							self.prepare_config_and_inputs()
            (
                (
                A_
            )     ,    (
                A_
            )     ,    (
                A_
            )     ,    (
                A_
            )     ,    (
                A_
            )     ,    (
                A_
            )     ,    (
                A_
            )     ,    (
                A_
            )     ,    (
                A_
            )     ,    
            )	     =							config_and_inputs
            A_	     =							{
                "input_ids": input_ids,
                "token_type_ids": token_type_ids,
                "langs": token_type_ids,
                "lengths": input_lengths,
            }
            return config, inputs_dict




@require_tf
class      _a				( snake_case_    ,							snake_case_    ,							unittest.TestCase							):




      """simple docstring"""

      _lowerCamelCase					: Optional[int]					=					(
          (
              TFXLMModel,
              TFXLMWithLMHeadModel,
              TFXLMForSequenceClassification,
              TFXLMForQuestionAnsweringSimple,
              TFXLMForTokenClassification,
              TFXLMForMultipleChoice,
          )
          if is_tf_available()
          else ()
      )
      _lowerCamelCase					: Optional[int]					=					(
          (TFXLMWithLMHeadModel,) if is_tf_available() else ()
      )  # TODO (PVP): Check other models whether language generation is also applicable
      _lowerCamelCase					: Optional[Any]					=					(
          {
              'feature-extraction': TFXLMModel,
              'fill-mask': TFXLMWithLMHeadModel,
              'question-answering': TFXLMForQuestionAnsweringSimple,
              'text-classification': TFXLMForSequenceClassification,
              'text-generation': TFXLMWithLMHeadModel,
              'token-classification': TFXLMForTokenClassification,
              'zero-shot': TFXLMForSequenceClassification,
          }
          if is_tf_available()
          else {}
      )
      _lowerCamelCase					: Dict					=					False
      _lowerCamelCase					: Dict					=					False





      def 		__A  (      self       :							int	, UpperCAmelCase       :							int	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							str	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							List[Any]  ):
            if (
                pipeline_test_casse_name == "QAPipelineTests"
                and tokenizer_name is not None
                and not tokenizer_name.endswith("Fast"  )
            ):
                  # `QAPipelineTests` fails for a few models when the slower tokenizer are used.
                  # (The slower tokenizers were never used for pipeline tests before the pipeline testing rework)
                  # TODO: check (and possibly fix) the `QAPipelineTests` with slower tokenizer
                  return True

            return False





      def 		__A  (      self       :							Optional[Any]  ):
            A_	     =							TFXLMModelTester(self  )
            A_	     =							ConfigTester(self	, config_class=UpperCAmelCase	, emb_dim=37  )





      def 		__A  (      self       :							Any  ):
            self.config_tester.run_common_tests()





      def 		__A  (      self       :							Dict  ):
            A_	     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_xlm_model(*UpperCAmelCase  )





      def 		__A  (      self       :							Union[str, Any]  ):
            A_	     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_xlm_lm_head(*UpperCAmelCase  )





      def 		__A  (      self       :							int  ):
            A_	     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_xlm_qa(*UpperCAmelCase  )





      def 		__A  (      self       :							str  ):
            A_	     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_xlm_sequence_classif(*UpperCAmelCase  )





      def 		__A  (      self       :							Optional[int]  ):
            A_	     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_xlm_for_token_classification(*UpperCAmelCase  )





      def 		__A  (      self       :							Dict  ):
            A_	     =							self.model_tester.prepare_config_and_inputs()
            self.model_tester.create_and_check_xlm_for_multiple_choice(*UpperCAmelCase  )







      @slow
      def 		__A  (      self       :							Union[str, Any]  ):
            for model_name in TF_XLM_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
                  A_	     =							TFXLMModel.from_pretrained(UpperCAmelCase  )
                  self.assertIsNotNone(UpperCAmelCase  )




@require_tf
class      _a				( unittest.TestCase							):




      """simple docstring"""

      @slow
      def 		__A  (      self       :							Optional[int]  ):
            A_	     =							TFXLMWithLMHeadModel.from_pretrained("xlm-mlm-en-2048"  )
            A_	     =							tf.convert_to_tensor([[14, 447]]	, dtype=tf.intaa  )  # the president
            A_	     =							[
                14,
                447,
                14,
                447,
                14,
                447,
                14,
                447,
                14,
                447,
                14,
                447,
                14,
                447,
                14,
                447,
                14,
                447,
                14,
                447,
            ]  # the president the president the president the president the president the president the president the president the president the president
            # TODO(PVP): this and other input_ids I tried for generation give pretty bad results. Not sure why. Model might just not be made for auto-regressive inference
            A_	     =							model.generate(UpperCAmelCase	, do_sample=UpperCAmelCase  )
            self.assertListEqual(output_ids[0].numpy().tolist()	, UpperCAmelCase  )


import json
import os
from typing import List, Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer
from ...utils import logging


__a						:Optional[Any]            = logging.get_logger(__name__)

__a						:Union[str, Any]            = {
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

__a						:Union[str, Any]            = {
    'vocab_file': {
        'microsoft/biogpt': 'https://huggingface.co/microsoft/biogpt/resolve/main/vocab.json',
    },
    'merges_file': {'microsoft/biogpt': 'https://huggingface.co/microsoft/biogpt/resolve/main/merges.txt'},
}

__a						:str            = {
    'microsoft/biogpt': 1024,
}

def 			__snake_case	(   __UpperCamelCase  :   Any						):



      """simple docstring"""

      A_	     =							set()
      A_	     =							word[0]
      for char in word[1:]:
            pairs.add((prev_char, char)						)
            A_	     =							char
      return pairs



class      _a				( snake_case_							):




      """simple docstring"""

      _lowerCamelCase					: Tuple					=					VOCAB_FILES_NAMES
      _lowerCamelCase					: Any					=					PRETRAINED_VOCAB_FILES_MAP
      _lowerCamelCase					: str					=					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      _lowerCamelCase					: List[Any]					=					['input_ids', 'attention_mask']





      def __init__(      self       :							Union[str, Any]	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							Optional[int]	, UpperCAmelCase       :							List[Any]="<unk>"	, UpperCAmelCase       :							Dict="<s>"	, UpperCAmelCase       :							Tuple="</s>"	, UpperCAmelCase       :							Optional[int]="</s>"	, UpperCAmelCase       :							int="<pad>"	, **UpperCAmelCase       :							List[str]	, ):
            super().__init__(
                bos_token=UpperCAmelCase	, eos_token=UpperCAmelCase	, sep_token=UpperCAmelCase	, unk_token=UpperCAmelCase	, pad_token=UpperCAmelCase	, **UpperCAmelCase	, )

            try:
                  import sacremoses
            except ImportError:
                  raise ImportError(
                      "You need to install sacremoses to use BioGptTokenizer. "
                      "See https://pypi.org/project/sacremoses/ for installation."  )

            A_	     =							"en"
            A_	     =							sacremoses
            # cache of sm.MosesTokenizer instance
            A_	     =							{}
            A_	     =							{}
            with open(UpperCAmelCase	, encoding="utf-8"  ) as vocab_handle:
                  A_	     =							json.load(UpperCAmelCase  )
            A_	     =							{v: k for k, v in self.encoder.items()}
            with open(UpperCAmelCase	, encoding="utf-8"  ) as merges_handle:
                  A_	     =							merges_handle.read().split("\n"  )[:-1]
            A_	     =							[tuple(merge.split()[:2]  ) for merge in merges]
            A_	     =							dict(zip(UpperCAmelCase	, range(len(UpperCAmelCase  )  )  )  )
            A_	     =							{}





      @property
      def 		__A  (      self       :							int  ):
            return len(self.encoder  )





      def 		__A  (      self       :							Union[str, Any]  ):
            return dict(self.encoder	, **self.added_tokens_encoder  )





      def 		__A  (      self       :							Any	, UpperCAmelCase       :							Union[str, Any]	, UpperCAmelCase       :							Optional[Any]  ):
            if lang not in self.cache_moses_tokenizer:
                  A_	     =							self.sm.MosesTokenizer(lang=UpperCAmelCase  )
                  A_	     =							moses_tokenizer
            return self.cache_moses_tokenizer[lang].tokenize(
                UpperCAmelCase	, aggressive_dash_splits=UpperCAmelCase	, return_str=UpperCAmelCase	, escape=UpperCAmelCase  )





      def 		__A  (      self       :							List[Any]	, UpperCAmelCase       :							Tuple	, UpperCAmelCase       :							List[Any]  ):
            if lang not in self.cache_moses_detokenizer:
                  A_	     =							self.sm.MosesDetokenizer(lang=UpperCAmelCase  )
                  A_	     =							moses_detokenizer
            return self.cache_moses_detokenizer[lang].detokenize(UpperCAmelCase  )





      def 		__A  (      self       :							Optional[Any]	, UpperCAmelCase       :							int  ):
            A_	     =							tuple(token[:-1]  ) + (token[-1] + "</w>",)
            if token in self.cache:
                  return self.cache[token]
            A_	     =							get_pairs(UpperCAmelCase  )

            if not pairs:
                  return token + "</w>"

            while True:
                  A_	     =							min(UpperCAmelCase	, key=lambda UpperCAmelCase  : self.bpe_ranks.get(UpperCAmelCase	, float("inf"  )  )  )
                  if bigram not in self.bpe_ranks:
                        break
                  A_     ,    A_	     =							bigram
                  A_	     =							[]
                  A_	     =							0
                  while i < len(UpperCAmelCase  ):
                        try:
                              A_	     =							word.index(UpperCAmelCase	, UpperCAmelCase  )
                        except ValueError:
                              new_word.extend(word[i:]  )
                              break
                        else:
                              new_word.extend(word[i:j]  )
                              A_	     =							j

                        if word[i] == first and i < len(UpperCAmelCase  ) - 1 and word[i + 1] == second:
                              new_word.append(first + second  )
                              i += 2
                        else:
                              new_word.append(word[i]  )
                              i += 1
                  A_	     =							tuple(UpperCAmelCase  )
                  A_	     =							new_word
                  if len(UpperCAmelCase  ) == 1:
                        break
                  else:
                        A_	     =							get_pairs(UpperCAmelCase  )
            A_	     =							" ".join(UpperCAmelCase  )
            if word == "\n  </w>":
                  A_	     =							"\n</w>"
            A_	     =							word
            return word





      def 		__A  (      self       :							Optional[int]	, UpperCAmelCase       :							str	, UpperCAmelCase       :							Optional[Any]=False  ):
            if bypass_tokenizer:
                  A_	     =							text.split()
            else:
                  A_	     =							self.moses_tokenize(UpperCAmelCase	, self.lang  )

            A_	     =							[]
            for token in text:
                  if token:
                        split_tokens.extend(list(self.bpe(UpperCAmelCase  ).split(" "  )  )  )

            return split_tokens





      def 		__A  (      self       :							Dict	, UpperCAmelCase       :							Union[str, Any]  ):
            return self.encoder.get(UpperCAmelCase	, self.encoder.get(self.unk_token  )  )





      def 		__A  (      self       :							Optional[Any]	, UpperCAmelCase       :							List[str]  ):
            return self.decoder.get(UpperCAmelCase	, self.unk_token  )





      def 		__A  (      self       :							Dict	, UpperCAmelCase       :							Any  ):
            A_	     =							[t.replace(" "	, ""  ).replace("</w>"	, " "  ) for t in tokens]
            A_	     =							"".join(UpperCAmelCase  ).split()
            # detokenize
            A_	     =							self.moses_detokenize(UpperCAmelCase	, self.lang  )
            return text





      def 		__A  (      self       :							List[str]	, UpperCAmelCase       :							List[int]	, UpperCAmelCase       :							Optional[List[int]] = None  ):
            if token_ids_a is None:
                  return [self.sep_token_id] + token_ids_a
            A_	     =							[self.sep_token_id]
            return sep + token_ids_a + sep + token_ids_a





      def 		__A  (      self       :							Any	, UpperCAmelCase       :							List[int]	, UpperCAmelCase       :							Optional[List[int]] = None	, UpperCAmelCase       :							bool = False  ):
            if already_has_special_tokens:
                  return super().get_special_tokens_mask(
                      token_ids_a=UpperCAmelCase	, token_ids_a=UpperCAmelCase	, already_has_special_tokens=UpperCAmelCase  )
            # no bos used in fairseq
            if token_ids_a is not None:
                  return [1] + ([0] * len(UpperCAmelCase  )) + [1] + ([0] * len(UpperCAmelCase  ))
            return [1] + ([0] * len(UpperCAmelCase  ))





      def 		__A  (      self       :							Optional[Any]	, UpperCAmelCase       :							List[int]	, UpperCAmelCase       :							Optional[List[int]] = None  ):
            A_	     =							[self.sep_token_id]

            # no bos used in fairseq
            if token_ids_a is None:
                  return len(token_ids_a + sep  ) * [0]
            return len(token_ids_a + sep  ) * [0] + len(token_ids_a + sep  ) * [1]





      def 		__A  (      self       :							Union[str, Any]	, UpperCAmelCase       :							str	, UpperCAmelCase       :							Optional[str] = None  ):
            if not os.path.isdir(UpperCAmelCase  ):
                  logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''  )
                  return
            A_	     =							os.path.join(
                UpperCAmelCase	, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]  )
            A_	     =							os.path.join(
                UpperCAmelCase	, (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]  )

            with open(UpperCAmelCase	, "w"	, encoding="utf-8"  ) as f:
                  f.write(json.dumps(self.encoder	, indent=2	, sort_keys=UpperCAmelCase	, ensure_ascii=UpperCAmelCase  ) + "\n"  )

            A_	     =							0
            with open(UpperCAmelCase	, "w"	, encoding="utf-8"  ) as writer:
                  for bpe_tokens, token_index in sorted(self.bpe_ranks.items()	, key=lambda UpperCAmelCase  : kv[1]  ):
                        if index != token_index:
                              logger.warning(
                                  f'''Saving vocabulary to {merge_file}: BPE merge indices are not consecutive.'''
                                  " Please check that the tokenizer is not corrupted!"  )
                              A_	     =							token_index
                        writer.write(" ".join(UpperCAmelCase  ) + "\n"  )
                        index += 1

            return vocab_file, merge_file





      def __getstate__(      self       :							Union[str, Any]  ):
            A_	     =							self.__dict__.copy()
            A_	     =							None
            return state







      def __setstate__(      self       :							List[Any]	, UpperCAmelCase       :							Dict  ):
            A_	     =							d

            try:
                  import sacremoses
            except ImportError:
                  raise ImportError(
                      "You need to install sacremoses to use XLMTokenizer. "
                      "See https://pypi.org/project/sacremoses/ for installation."  )

            A_	     =							sacremoses