





import unittest

from transformers import is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
     import torch

     from transformers import (
         OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST,
         OpenAIGPTConfig,
         OpenAIGPTDoubleHeadsModel,
         OpenAIGPTForSequenceClassification,
         OpenAIGPTLMHeadModel,
         OpenAIGPTModel,
     )



class 							_UpperCAmelCase       :

  """simple docstring"""


  def __init__(							self					:						Any   ,    lowerCAmelCase_					:						int   ,    lowerCAmelCase_					:						Any=1_3   ,    lowerCAmelCase_					:						Dict=7   ,    lowerCAmelCase_					:						Optional[int]=True   ,    lowerCAmelCase_					:						Any=True   ,    lowerCAmelCase_					:						Union[str, Any]=True   ,    lowerCAmelCase_					:						str=9_9   ,    lowerCAmelCase_					:						List[str]=3_2   ,    lowerCAmelCase_					:						Optional[int]=5   ,    lowerCAmelCase_					:						Dict=4   ,    lowerCAmelCase_					:						str=3_7   ,    lowerCAmelCase_					:						Union[str, Any]="gelu"   ,    lowerCAmelCase_					:						Any=0.1   ,    lowerCAmelCase_					:						str=0.1   ,    lowerCAmelCase_					:						Optional[int]=5_1_2   ,    lowerCAmelCase_					:						Optional[Any]=1_6   ,    lowerCAmelCase_					:						int=2   ,    lowerCAmelCase_					:						Tuple=0.02   ,    lowerCAmelCase_					:						str=3   ,    lowerCAmelCase_					:						Union[str, Any]=4   ,    lowerCAmelCase_					:						List[str]=None   ,    )				->		Union[str, Any]:
     __lowerCAmelCase				      =      parent
     __lowerCAmelCase				      =      batch_size
     __lowerCAmelCase				      =      seq_length
     __lowerCAmelCase				      =      is_training
     __lowerCAmelCase				      =      use_token_type_ids
     __lowerCAmelCase				      =      use_labels
     __lowerCAmelCase				      =      vocab_size
     __lowerCAmelCase				      =      hidden_size
     __lowerCAmelCase				      =      num_hidden_layers
     __lowerCAmelCase				      =      num_attention_heads
     __lowerCAmelCase				      =      intermediate_size
     __lowerCAmelCase				      =      hidden_act
     __lowerCAmelCase				      =      hidden_dropout_prob
     __lowerCAmelCase				      =      attention_probs_dropout_prob
     __lowerCAmelCase				      =      max_position_embeddings
     __lowerCAmelCase				      =      type_vocab_size
     __lowerCAmelCase				      =      type_sequence_label_size
     __lowerCAmelCase				      =      initializer_range
     __lowerCAmelCase				      =      num_labels
     __lowerCAmelCase				      =      num_choices
     __lowerCAmelCase				      =      scope
     __lowerCAmelCase				      =      self.vocab_size - 1






  def 							lowercase						(							self					:						Dict	)				->		Tuple:
     __lowerCAmelCase				      =      ids_tensor([self.batch_size, self.seq_length]   ,    self.vocab_size	)

     __lowerCAmelCase				      =      None
     if self.use_token_type_ids:
        __lowerCAmelCase				      =      ids_tensor([self.batch_size, self.seq_length]   ,    self.type_vocab_size	)

     __lowerCAmelCase				      =      None
     __lowerCAmelCase				      =      None
     __lowerCAmelCase				      =      None
     if self.use_labels:
        __lowerCAmelCase				      =      ids_tensor([self.batch_size]   ,    self.type_sequence_label_size	)
        __lowerCAmelCase				      =      ids_tensor([self.batch_size, self.seq_length]   ,    self.num_labels	)
        __lowerCAmelCase				      =      ids_tensor([self.batch_size]   ,    self.num_choices	)

     __lowerCAmelCase				      =      OpenAIGPTConfig(
         vocab_size=self.vocab_size   ,    n_embd=self.hidden_size   ,    n_layer=self.num_hidden_layers   ,    n_head=self.num_attention_heads   ,    n_positions=self.max_position_embeddings   ,    pad_token_id=self.pad_token_id   ,    )

     __lowerCAmelCase				      =      ids_tensor([self.num_hidden_layers, self.num_attention_heads]   ,    2	)

     return (
         config,
         input_ids,
         head_mask,
         token_type_ids,
         sequence_labels,
         token_labels,
         choice_labels,
     )






  def 							lowercase						(							self					:						str   ,    lowerCAmelCase_					:						List[str]   ,    lowerCAmelCase_					:						Any   ,    lowerCAmelCase_					:						Tuple   ,    lowerCAmelCase_					:						int   ,    *lowerCAmelCase_					:						str	)				->		List[str]:
     __lowerCAmelCase				      =      OpenAIGPTModel(config=_lowerCAmelCase	)
     model.to(_lowerCAmelCase	)
     model.eval()

     __lowerCAmelCase				      =      model(_lowerCAmelCase   ,    token_type_ids=_lowerCAmelCase   ,    head_mask=_lowerCAmelCase	)
     __lowerCAmelCase				      =      model(_lowerCAmelCase   ,    token_type_ids=_lowerCAmelCase	)
     __lowerCAmelCase				      =      model(_lowerCAmelCase	)

     self.parent.assertEqual(result.last_hidden_state.shape   ,    (self.batch_size, self.seq_length, self.hidden_size)	)






  def 							lowercase						(							self					:						str   ,    lowerCAmelCase_					:						int   ,    lowerCAmelCase_					:						Tuple   ,    lowerCAmelCase_					:						List[Any]   ,    lowerCAmelCase_					:						Optional[Any]   ,    *lowerCAmelCase_					:						Dict	)				->		List[str]:
     __lowerCAmelCase				      =      OpenAIGPTLMHeadModel(_lowerCAmelCase	)
     model.to(_lowerCAmelCase	)
     model.eval()

     __lowerCAmelCase				      =      model(_lowerCAmelCase   ,    token_type_ids=_lowerCAmelCase   ,    labels=_lowerCAmelCase	)
     self.parent.assertEqual(result.loss.shape   ,    ()	)
     self.parent.assertEqual(result.logits.shape   ,    (self.batch_size, self.seq_length, self.vocab_size)	)






  def 							lowercase						(							self					:						List[str]   ,    lowerCAmelCase_					:						Optional[Any]   ,    lowerCAmelCase_					:						Tuple   ,    lowerCAmelCase_					:						Dict   ,    lowerCAmelCase_					:						Tuple   ,    *lowerCAmelCase_					:						List[Any]	)				->		List[Any]:
     __lowerCAmelCase				      =      OpenAIGPTDoubleHeadsModel(_lowerCAmelCase	)
     model.to(_lowerCAmelCase	)
     model.eval()

     __lowerCAmelCase				      =      model(_lowerCAmelCase   ,    token_type_ids=_lowerCAmelCase   ,    labels=_lowerCAmelCase	)
     self.parent.assertEqual(result.loss.shape   ,    ()	)
     self.parent.assertEqual(result.logits.shape   ,    (self.batch_size, self.seq_length, self.vocab_size)	)






  def 							lowercase						(							self					:						List[Any]   ,    lowerCAmelCase_					:						Any   ,    lowerCAmelCase_					:						Dict   ,    lowerCAmelCase_					:						Tuple   ,    lowerCAmelCase_					:						Tuple   ,    *lowerCAmelCase_					:						Optional[Any]	)				->		List[Any]:
     __lowerCAmelCase				      =      self.num_labels
     __lowerCAmelCase				      =      OpenAIGPTForSequenceClassification(_lowerCAmelCase	)
     model.to(_lowerCAmelCase	)
     model.eval()

     __lowerCAmelCase				      =      ids_tensor([self.batch_size]   ,    self.type_sequence_label_size	)
     __lowerCAmelCase				      =      model(_lowerCAmelCase   ,    token_type_ids=_lowerCAmelCase   ,    labels=_lowerCAmelCase	)
     self.parent.assertEqual(result.logits.shape   ,    (self.batch_size, self.num_labels)	)






  def 							lowercase						(							self					:						Optional[Any]	)				->		Tuple:
     __lowerCAmelCase				      =      self.prepare_config_and_inputs()
     (
         (
         __lowerCAmelCase     
     )      ,      (
         __lowerCAmelCase     
     )      ,      (
         __lowerCAmelCase     
     )      ,      (
         __lowerCAmelCase     
     )      ,      (
         __lowerCAmelCase     
     )      ,      (
         __lowerCAmelCase     
     )      ,      (
         __lowerCAmelCase     
     )      ,           
     )				      =      config_and_inputs
     __lowerCAmelCase				      =      {
         'input_ids': input_ids,
         'token_type_ids': token_type_ids,
         'head_mask': head_mask,
     }

     return config, inputs_dict





@require_torch
class 							_UpperCAmelCase       (       _UpperCamelCase						,       _UpperCamelCase						,       _UpperCamelCase						,       unittest.TestCase       ):

  """simple docstring"""


  a_              =					(
      (OpenAIGPTModel, OpenAIGPTLMHeadModel, OpenAIGPTDoubleHeadsModel, OpenAIGPTForSequenceClassification)
      if is_torch_available()
      else ()
  )
  a_              =					(
      (OpenAIGPTLMHeadModel,) if is_torch_available() else ()
  )  # TODO (PVP): Add Double HeadsModel when generate() function is changed accordingly
  a_              =					(
      {
          """feature-extraction""": OpenAIGPTModel,
          """text-classification""": OpenAIGPTForSequenceClassification,
          """text-generation""": OpenAIGPTLMHeadModel,
          """zero-shot""": OpenAIGPTForSequenceClassification,
      }
      if is_torch_available()
      else {}
  )






  def 							lowercase						(							self					:						Optional[int]   ,    lowerCAmelCase_					:						Optional[int]   ,    lowerCAmelCase_					:						Optional[Any]   ,    lowerCAmelCase_					:						List[str]   ,    lowerCAmelCase_					:						Dict   ,    lowerCAmelCase_					:						List[Any]	)				->		Dict:
     if pipeline_test_casse_name == "ZeroShotClassificationPipelineTests":
        # Get `tokenizer does not have a padding token` error for both fast/slow tokenizers.
        # `OpenAIGPTConfig` was never used in pipeline tests, either because of a missing checkpoint or because a
        # tiny config could not be created.
        return True

     return False






  def 							lowercase						(							self					:						Optional[Any]   ,    lowerCAmelCase_					:						Optional[int]   ,    lowerCAmelCase_					:						Optional[Any]   ,    lowerCAmelCase_					:						List[Any]=False	)				->		int:
     __lowerCAmelCase				      =      super()._prepare_for_class(_lowerCAmelCase   ,    _lowerCAmelCase   ,    return_labels=_lowerCAmelCase	)

     if return_labels:
        if model_class.__name__ == "OpenAIGPTDoubleHeadsModel":
           __lowerCAmelCase				      =      torch.zeros(
               (self.model_tester.batch_size, self.model_tester.num_choices, self.model_tester.seq_length)   ,    dtype=torch.long   ,    device=_lowerCAmelCase   ,    )
           __lowerCAmelCase				      =      inputs_dict['labels']
           __lowerCAmelCase				      =      inputs_dict['labels']
           __lowerCAmelCase				      =      torch.zeros(
               (self.model_tester.batch_size, self.model_tester.num_choices)   ,    dtype=torch.long   ,    device=_lowerCAmelCase   ,    )
           __lowerCAmelCase				      =      torch.zeros(
               self.model_tester.batch_size   ,    dtype=torch.long   ,    device=_lowerCAmelCase	)
     return inputs_dict






  def 							lowercase						(							self					:						str	)				->		Union[str, Any]:
     __lowerCAmelCase				      =      OpenAIGPTModelTester(self	)
     __lowerCAmelCase				      =      ConfigTester(self   ,    config_class=_lowerCAmelCase   ,    n_embd=3_7	)






  def 							lowercase						(							self					:						List[str]	)				->		List[str]:
     self.config_tester.run_common_tests()






  def 							lowercase						(							self					:						List[Any]	)				->		Optional[int]:
     __lowerCAmelCase				      =      self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_openai_gpt_model(*_lowerCAmelCase	)






  def 							lowercase						(							self					:						Dict	)				->		Optional[Any]:
     __lowerCAmelCase				      =      self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_lm_head_model(*_lowerCAmelCase	)






  def 							lowercase						(							self					:						Any	)				->		List[Any]:
     __lowerCAmelCase				      =      self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_double_lm_head_model(*_lowerCAmelCase	)






  def 							lowercase						(							self					:						Any	)				->		int:
     __lowerCAmelCase				      =      self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_openai_gpt_for_sequence_classification(*_lowerCAmelCase	)






  @slow
  def 							lowercase						(							self					:						int	)				->		Union[str, Any]:
     for model_name in OPENAI_GPT_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
        __lowerCAmelCase				      =      OpenAIGPTModel.from_pretrained(_lowerCAmelCase	)
        self.assertIsNotNone(_lowerCAmelCase	)





@require_torch
class 							_UpperCAmelCase       (       unittest.TestCase       ):

  """simple docstring"""


  @slow
  def 							lowercase						(							self					:						str	)				->		Union[str, Any]:
     __lowerCAmelCase				      =      OpenAIGPTLMHeadModel.from_pretrained('openai-gpt'	)
     model.to(_lowerCAmelCase	)
     __lowerCAmelCase				      =      torch.tensor([[4_8_1, 4_7_3_5, 5_4_4]]   ,    dtype=torch.long   ,    device=_lowerCAmelCase	)  # the president is
     __lowerCAmelCase				      =      [
         4_8_1,
         4_7_3_5,
         5_4_4,
         2_4_6,
         9_6_3,
         8_7_0,
         7_6_2,
         2_3_9,
         2_4_4,
         4_0_4_7_7,
         2_4_4,
         2_4_9,
         7_1_9,
         8_8_1,
         4_8_7,
         5_4_4,
         2_4_0,
         2_4_4,
         6_0_3,
         4_8_1,
     ]  # the president is a very good man. " \n " i\'m sure he is, " said the

     __lowerCAmelCase				      =      model.generate(_lowerCAmelCase   ,    do_sample=_lowerCAmelCase	)
     self.assertListEqual(output_ids[0].tolist()   ,    _lowerCAmelCase	)





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

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


SCREAMING_SNAKE_CASE						:Union[str, Any]			=  logging.get_logger(__name__)


SCREAMING_SNAKE_CASE						:Dict			=  {
    '''vocab_file''': '''vocab.json''',
    '''tokenizer_config_file''': '''tokenizer_config.json''',
    '''merges_file''': '''merges.txt''',
}

SCREAMING_SNAKE_CASE						:Optional[int]			=  {
    '''vocab_file''': {
        '''facebook/s2t-wav2vec2-large-en-de''': (
            '''https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/vocab.json'''
        ),
    },
    '''tokenizer_config_file''': {
        '''facebook/s2t-wav2vec2-large-en-de''': (
            '''https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/tokenizer_config.json'''
        ),
    },
    '''merges_file''': {
        '''facebook/s2t-wav2vec2-large-en-de''': (
            '''https://huggingface.co/facebook/s2t-wav2vec2-large-en-de/resolve/main/merges.txt'''
        ),
    },
}

SCREAMING_SNAKE_CASE						:List[str]			=  '''</w>'''
SCREAMING_SNAKE_CASE						:int			=  '''@@ '''



def       _lowerCAmelCase    (				lowerCAmelCase_				:int				)->int:
   '''simple docstring'''






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


# Speech2Text2 has no max input length
SCREAMING_SNAKE_CASE						:Tuple			=  {'''facebook/s2t-wav2vec2-large-en-de''': 10_24}



class      __lowerCAmelCase  (			a       ):






  """simple docstring"""





  _SCREAMING_SNAKE_CASE		     =							VOCAB_FILES_NAMES
  _SCREAMING_SNAKE_CASE		     =							PRETRAINED_VOCAB_FILES_MAP
  _SCREAMING_SNAKE_CASE		     =							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
  _SCREAMING_SNAKE_CASE		     =							['input_ids', 'attention_mask']



  def __init__(  self							:							List[Any]     ,					_lowerCAmelCase							:							str     ,					_lowerCAmelCase							:							Union[str, Any]="<s>"     ,					_lowerCAmelCase							:							Tuple="<pad>"     ,					_lowerCAmelCase							:							Union[str, Any]="</s>"     ,					_lowerCAmelCase							:							str="<unk>"     ,					_lowerCAmelCase							:							Union[str, Any]=False     ,					_lowerCAmelCase							:							List[str]=None     ,					**_lowerCAmelCase							:							List[str]     ,					) ->			List[str]:




     """simple docstring"""
     super().__init__(
         unk_token=_lowerCAmelCase     ,					bos_token=_lowerCAmelCase     ,					eos_token=_lowerCAmelCase     ,					pad_token=_lowerCAmelCase     ,					do_lower_case=_lowerCAmelCase     ,					**_lowerCAmelCase     ,					)

     snake_case_            =       do_lower_case

     with open(_lowerCAmelCase     ,					encoding="utf-8"	) as vocab_handle:
        snake_case_            =       json.load(_lowerCAmelCase	)
     snake_case_            =       {v: k for k, v in self.encoder.items()}

     if merges_file is None:
        logger.info(F'''No merges files provided. {self.__class__.__name__} can only be used for decoding.'''	)

        snake_case_            =       None
        snake_case_            =       None
     else:
        with open(_lowerCAmelCase     ,					encoding="utf-8"	) as merges_handle:
           snake_case_            =       merges_handle.read().split("\n"	)[:-1]

        snake_case_            =       [tuple(merge.split()[:2]	) for merge in merges]
        snake_case_            =       dict(zip(_lowerCAmelCase     ,					range(len(_lowerCAmelCase	)	)	)	)
        snake_case_            =       {}



  @property
  def        lowerCAmelCase__	(  self							:							Union[str, Any]	) ->			int:




     """simple docstring"""
     return len(self.decoder	)



  def        lowerCAmelCase__	(  self							:							Any	) ->			Dict:




     """simple docstring"""
     return dict(self.encoder     ,					**self.added_tokens_encoder	)



  def        lowerCAmelCase__	(  self							:							Union[str, Any]     ,					_lowerCAmelCase							:							Any	) ->			Tuple:




     """simple docstring"""
     snake_case_            =       tuple(token[:-1]	) + (token[-1] + BPE_TOKEN_MERGES,)
     if token in self.cache:
        return self.cache[token]
     snake_case_            =       get_pairs(_lowerCAmelCase	)

     if not pairs:
        return token

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

           if word[i] == first and i < len(_lowerCAmelCase	) - 1 and word[i + 1] == second:
              new_word.append(first + second	)
              i += 2
           else:
              new_word.append(word[i]	)
              i += 1
        snake_case_            =       tuple(_lowerCAmelCase	)
        snake_case_            =       new_word
        if len(_lowerCAmelCase	) == 1:
           break
        else:
           snake_case_            =       get_pairs(_lowerCAmelCase	)
     snake_case_            =       " ".join(_lowerCAmelCase	)
     if word == "\n  " + BPE_TOKEN_MERGES:
        snake_case_            =       "\n" + BPE_TOKEN_MERGES

     if word.endswith(_lowerCAmelCase	):
        snake_case_            =       word.replace(_lowerCAmelCase     ,					""	)

     snake_case_            =       word.replace(" "     ,					_lowerCAmelCase	)
     snake_case_            =       word
     return word



  def        lowerCAmelCase__	(  self							:							List[Any]     ,					_lowerCAmelCase							:							List[str]	) ->			List[Any]:




     """simple docstring"""

     if self.bpe_ranks is None:
        raise ValueError(
            "This tokenizer was instantiated without a `merges.txt` file, so"
            " that it can only be used for decoding, not for encoding."
            "Make sure to provide `merges.txt` file at instantiation to enable "
            "encoding."	)

     if self.do_lower_case:
        snake_case_            =       text.lower()

     snake_case_            =       text.split()

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

     return split_tokens



  def        lowerCAmelCase__	(  self							:							str     ,					_lowerCAmelCase							:							str	) ->			int:




     """simple docstring"""
     return self.encoder.get(_lowerCAmelCase     ,					self.encoder.get(self.unk_token	)	)



  def        lowerCAmelCase__	(  self							:							int     ,					_lowerCAmelCase							:							int	) ->			str:




     """simple docstring"""
     snake_case_            =       self.decoder.get(_lowerCAmelCase     ,					self.unk_token	)
     return result



  def        lowerCAmelCase__	(  self							:							Union[str, Any]     ,					_lowerCAmelCase							:							List[str]	) ->			str:




     """simple docstring"""
     snake_case_            =       " ".join(_lowerCAmelCase	)

     # make sure @@ tokens are concatenated
     snake_case_            =       "".join(string.split(_lowerCAmelCase	)	)

     return string



  def        lowerCAmelCase__	(  self							:							Tuple     ,					_lowerCAmelCase							:							str     ,					_lowerCAmelCase							:							Optional[str] = None	) ->			Tuple[str]:




     """simple docstring"""
     if not os.path.isdir(_lowerCAmelCase	):
        logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''	)
        return
     snake_case_            =       os.path.join(
         _lowerCAmelCase     ,					(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]	)
     snake_case_            =       os.path.join(
         _lowerCAmelCase     ,					(filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["merges_file"]	)

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

     snake_case_            =       0
     if self.bpe_ranks is None:
        return (vocab_file,)

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

     return (vocab_file, merges_file)




