





'''simple docstring'''


import tempfile
import unittest

from transformers import SPIECE_UNDERLINE, BatchEncoding, PLBartTokenizer, is_torch_available
from transformers.testing_utils import (
    get_tests_dir,
    nested_simplify,
    require_sentencepiece,
    require_tokenizers,
    require_torch,
)

from ...test_tokenization_common import TokenizerTesterMixin


_SCREAMING_SNAKE_CASE	 =							get_tests_dir("fixtures/test_sentencepiece.model")


if is_torch_available():
       from transformers.models.plbart.modeling_plbart import shift_tokens_right

_SCREAMING_SNAKE_CASE	 =							5_00_03
_SCREAMING_SNAKE_CASE	 =							5_00_02

@require_sentencepiece
@require_tokenizers
class    lowerCAmelCase_		(		__magic_name__						,unittest.TestCase		):
 __lowerCamelCase						:							int         =					PLBartTokenizer
 __lowerCamelCase						:							Optional[Any]         =					None
 __lowerCamelCase						:							Union[str, Any]         =					False




 def 							_snake_case   (	self		)    ->							Optional[int]:
     super().setUp()

     # We have a SentencePiece fixture for testing
     _lowerCAmelCase				=      PLBartTokenizer(_lowerCAmelCase			,     language_codes="base"			,     keep_accents=_lowerCAmelCase		)
     tokenizer.save_pretrained(self.tmpdirname		)




 def 							_snake_case   (	self		)    ->							Union[str, Any]:
     _lowerCAmelCase				=      PLBartTokenizer(_lowerCAmelCase			,     language_codes="base"			,     keep_accents=_lowerCAmelCase		)

     _lowerCAmelCase				=      tokenizer.tokenize("This is a test"		)
     self.assertListEqual(_lowerCAmelCase			,     ["▁This", "▁is", "▁a", "▁t", "est"]		)

     self.assertListEqual(
         tokenizer.convert_tokens_to_ids(_lowerCAmelCase		)			,     [value + tokenizer.fairseq_offset for value in [285, 46, 10, 170, 382]]			,     )

     _lowerCAmelCase				=      tokenizer.tokenize("I was born in 92000, and this is falsé."		)
     self.assertListEqual(
         _lowerCAmelCase			,     [
             SPIECE_UNDERLINE + "I",
             SPIECE_UNDERLINE + "was",
             SPIECE_UNDERLINE + "b",
             "or",
             "n",
             SPIECE_UNDERLINE + "in",
             SPIECE_UNDERLINE + "",
             "9",
             "2",
             "0",
             "0",
             "0",
             ",",
             SPIECE_UNDERLINE + "and",
             SPIECE_UNDERLINE + "this",
             SPIECE_UNDERLINE + "is",
             SPIECE_UNDERLINE + "f",
             "al",
             "s",
             "é",
             ".",
         ]			,     )
     _lowerCAmelCase				=      tokenizer.convert_tokens_to_ids(_lowerCAmelCase		)
     self.assertListEqual(
         _lowerCAmelCase			,     [
             value + tokenizer.fairseq_offset
             for value in [8, 21, 84, 55, 24, 19, 7, 2, 602, 347, 347, 347, 3, 12, 66, 46, 72, 80, 6, 2, 4]
         ]			,     )

     _lowerCAmelCase				=      tokenizer.convert_ids_to_tokens(_lowerCAmelCase		)
     self.assertListEqual(
         _lowerCAmelCase			,     [
             SPIECE_UNDERLINE + "I",
             SPIECE_UNDERLINE + "was",
             SPIECE_UNDERLINE + "b",
             "or",
             "n",
             SPIECE_UNDERLINE + "in",
             SPIECE_UNDERLINE + "",
             "<unk>",
             "2",
             "0",
             "0",
             "0",
             ",",
             SPIECE_UNDERLINE + "and",
             SPIECE_UNDERLINE + "this",
             SPIECE_UNDERLINE + "is",
             SPIECE_UNDERLINE + "f",
             "al",
             "s",
             "<unk>",
             ".",
         ]			,     )

     _lowerCAmelCase				=      tokenizer.vocab_size
     _lowerCAmelCase				=      [tokenizer.convert_ids_to_tokens(_lowerCAmelCase		) for x in range(end - 4			,     _lowerCAmelCase		)]

     self.assertListEqual(_lowerCAmelCase			,     ["__java__", "__python__", "__en_XX__", "<mask>"]		)

     _lowerCAmelCase				=      "java.lang.Exception, python.lang.Exception, javascript, php, ruby, go"
     _lowerCAmelCase				=      tokenizer(_lowerCAmelCase		).input_ids
     self.assertEqual(
         tokenizer.decode(_lowerCAmelCase			,     skip_special_tokens=_lowerCAmelCase			,     clean_up_tokenization_spaces=_lowerCAmelCase		)			,     _lowerCAmelCase			,     )





 def 							_snake_case   (	self		)    ->							Dict:
     _lowerCAmelCase				=      PLBartTokenizer(_lowerCAmelCase			,     language_codes="multi"			,     keep_accents=_lowerCAmelCase		)

     _lowerCAmelCase				=      tokenizer.tokenize("This is a test"		)
     self.assertListEqual(_lowerCAmelCase			,     ["▁This", "▁is", "▁a", "▁t", "est"]		)

     self.assertListEqual(
         tokenizer.convert_tokens_to_ids(_lowerCAmelCase		)			,     [value + tokenizer.fairseq_offset for value in [285, 46, 10, 170, 382]]			,     )

     _lowerCAmelCase				=      tokenizer.tokenize("I was born in 92000, and this is falsé."		)
     self.assertListEqual(
         _lowerCAmelCase			,     [
             SPIECE_UNDERLINE + "I",
             SPIECE_UNDERLINE + "was",
             SPIECE_UNDERLINE + "b",
             "or",
             "n",
             SPIECE_UNDERLINE + "in",
             SPIECE_UNDERLINE + "",
             "9",
             "2",
             "0",
             "0",
             "0",
             ",",
             SPIECE_UNDERLINE + "and",
             SPIECE_UNDERLINE + "this",
             SPIECE_UNDERLINE + "is",
             SPIECE_UNDERLINE + "f",
             "al",
             "s",
             "é",
             ".",
         ]			,     )
     _lowerCAmelCase				=      tokenizer.convert_tokens_to_ids(_lowerCAmelCase		)
     self.assertListEqual(
         _lowerCAmelCase			,     [
             value + tokenizer.fairseq_offset
             for value in [8, 21, 84, 55, 24, 19, 7, 2, 602, 347, 347, 347, 3, 12, 66, 46, 72, 80, 6, 2, 4]
         ]			,     )

     _lowerCAmelCase				=      tokenizer.convert_ids_to_tokens(_lowerCAmelCase		)
     self.assertListEqual(
         _lowerCAmelCase			,     [
             SPIECE_UNDERLINE + "I",
             SPIECE_UNDERLINE + "was",
             SPIECE_UNDERLINE + "b",
             "or",
             "n",
             SPIECE_UNDERLINE + "in",
             SPIECE_UNDERLINE + "",
             "<unk>",
             "2",
             "0",
             "0",
             "0",
             ",",
             SPIECE_UNDERLINE + "and",
             SPIECE_UNDERLINE + "this",
             SPIECE_UNDERLINE + "is",
             SPIECE_UNDERLINE + "f",
             "al",
             "s",
             "<unk>",
             ".",
         ]			,     )
     _lowerCAmelCase				=      tokenizer.vocab_size
     _lowerCAmelCase				=      [tokenizer.convert_ids_to_tokens(_lowerCAmelCase		) for x in range(end - 7			,     _lowerCAmelCase		)]

     self.assertListEqual(
         _lowerCAmelCase			,     ["__java__", "__python__", "__en_XX__", "__javascript__", "__php__", "__ruby__", "__go__"]		)
     _lowerCAmelCase				=      "java.lang.Exception, python.lang.Exception, javascript, php, ruby, go"
     _lowerCAmelCase				=      tokenizer(_lowerCAmelCase		).input_ids
     self.assertEqual(
         tokenizer.decode(_lowerCAmelCase			,     skip_special_tokens=_lowerCAmelCase			,     clean_up_tokenization_spaces=_lowerCAmelCase		)			,     _lowerCAmelCase			,     )







@require_torch
@require_sentencepiece
@require_tokenizers
class    lowerCAmelCase_		(		unittest.TestCase		):
 __lowerCamelCase						:							Union[str, Any]         =					"uclanlp/plbart-python-en_XX"
 __lowerCamelCase						:							Any         =					[
     "def maximum(a,b,c):NEW_LINE_INDENTreturn max([a,b,c])",
     "def sum(a,b,c):NEW_LINE_INDENTreturn sum([a,b,c])",
 ]
 __lowerCamelCase						:							Optional[Any]         =					[
     "Returns the maximum value of a b c.",
     "Sums the values of a b c.",
 ]
 __lowerCamelCase						:							Optional[int]         =					[
     134,
     5_452,
     33_460,
     33_441,
     33_463,
     33_465,
     33_463,
     33_449,
     988,
     20,
     33_456,
     19,
     33_456,
     771,
     39,
     4_258,
     889,
     3_318,
     33_441,
     33_463,
     33_465,
     33_463,
     33_449,
     2_471,
     2,
     PYTHON_CODE,
 ]




 @classmethod
 def 							_snake_case   (	cls		)    ->							int:
     _lowerCAmelCase				=      PLBartTokenizer.from_pretrained(
         cls.checkpoint_name			,     language_codes="base"			,     src_lang="python"			,     tgt_lang="en_XX"		)
     _lowerCAmelCase				=      1
     return cls




 def 							_snake_case   (	self		)    ->							int:
     self.assertEqual(self.tokenizer.fairseq_tokens_to_ids["__java__"]			,     50001		)
     self.assertEqual(self.tokenizer.fairseq_tokens_to_ids["__python__"]			,     50002		)
     self.assertEqual(self.tokenizer.fairseq_tokens_to_ids["__en_XX__"]			,     50003		)




 def 							_snake_case   (	self		)    ->							List[str]:
     _lowerCAmelCase				=      self.tokenizer.batch_encode_plus(self.src_text		).input_ids[0]
     self.assertListEqual(self.expected_src_tokens			,     _lowerCAmelCase		)




 def 							_snake_case   (	self		)    ->							Optional[Any]:
     self.assertIn(_lowerCAmelCase			,     self.tokenizer.all_special_ids		)
     _lowerCAmelCase				=      [EN_CODE, 9037, 33442, 57, 752, 153, 14, 56, 18, 9, 2]
     _lowerCAmelCase				=      self.tokenizer.decode(_lowerCAmelCase			,     skip_special_tokens=_lowerCAmelCase		)
     _lowerCAmelCase				=      self.tokenizer.decode(generated_ids[1:]			,     skip_special_tokens=_lowerCAmelCase		)
     self.assertEqual(_lowerCAmelCase			,     _lowerCAmelCase		)
     self.assertNotIn(self.tokenizer.eos_token			,     _lowerCAmelCase		)




 def 							_snake_case   (	self		)    ->							int:
     _lowerCAmelCase				=      ["def sum(a,b,c):NEW_LINE_INDENTreturn sum([a,b,c])" * 20]
     self.assertIsInstance(src_text[0]			,     _lowerCAmelCase		)
     _lowerCAmelCase				=      10
     _lowerCAmelCase				=      self.tokenizer(_lowerCAmelCase			,     max_length=_lowerCAmelCase			,     truncation=_lowerCAmelCase		).input_ids[0]
     self.assertEqual(ids[-2]			,     2		)
     self.assertEqual(ids[-1]			,     _lowerCAmelCase		)
     self.assertEqual(len(_lowerCAmelCase		)			,     _lowerCAmelCase		)




 def 							_snake_case   (	self		)    ->							List[Any]:
     self.assertListEqual(self.tokenizer.convert_tokens_to_ids(["<mask>", "__java__"]		)			,     [50004, 50001]		)




 def 							_snake_case   (	self		)    ->							Tuple:
     _lowerCAmelCase				=      tempfile.mkdtemp()
     _lowerCAmelCase				=      self.tokenizer.fairseq_tokens_to_ids
     self.tokenizer.save_pretrained(_lowerCAmelCase		)
     _lowerCAmelCase				=      PLBartTokenizer.from_pretrained(_lowerCAmelCase		)
     self.assertDictEqual(new_tok.fairseq_tokens_to_ids			,     _lowerCAmelCase		)




 @require_torch
 def 							_snake_case   (	self		)    ->							Tuple:
     _lowerCAmelCase				=      self.tokenizer(self.src_text			,     text_target=self.tgt_text			,     padding=_lowerCAmelCase			,     return_tensors="pt"		)
     _lowerCAmelCase				=      shift_tokens_right(batch["labels"]			,     self.tokenizer.pad_token_id		)

     # fairseq batch: https://gist.github.com/sshleifer/cba08bc2109361a74ac3760a7e30e4f4
     self.assertEqual(batch.input_ids[1][-2:].tolist()			,     [2, PYTHON_CODE]		)
     self.assertEqual(batch.decoder_input_ids[1][0]			,     _lowerCAmelCase		)
     self.assertEqual(batch.decoder_input_ids[1][-1]			,     2		)
     self.assertEqual(batch.labels[1][-2:].tolist()			,     [2, EN_CODE]		)




 @require_torch
 def 							_snake_case   (	self		)    ->							Any:
     _lowerCAmelCase				=      self.tokenizer(
         self.src_text			,     text_target=self.tgt_text			,     padding=_lowerCAmelCase			,     truncation=_lowerCAmelCase			,     max_length=len(self.expected_src_tokens		)			,     return_tensors="pt"			,     )
     _lowerCAmelCase				=      shift_tokens_right(batch["labels"]			,     self.tokenizer.pad_token_id		)

     self.assertIsInstance(_lowerCAmelCase			,     _lowerCAmelCase		)

     self.assertEqual((2, 26)			,     batch.input_ids.shape		)
     self.assertEqual((2, 26)			,     batch.attention_mask.shape		)
     _lowerCAmelCase				=      batch.input_ids.tolist()[0]
     self.assertListEqual(self.expected_src_tokens			,     _lowerCAmelCase		)
     self.assertEqual(2			,     batch.decoder_input_ids[0, -1]		)  # EOS
     # Test that special tokens are reset
     self.assertEqual(self.tokenizer.prefix_tokens			,     []		)
     self.assertEqual(self.tokenizer.suffix_tokens			,     [self.tokenizer.eos_token_id, PYTHON_CODE]		)




 def 							_snake_case   (	self		)    ->							str:
     _lowerCAmelCase				=      self.tokenizer(self.src_text			,     padding=_lowerCAmelCase			,     truncation=_lowerCAmelCase			,     max_length=3			,     return_tensors="pt"		)
     _lowerCAmelCase				=      self.tokenizer(
         text_target=self.tgt_text			,     padding=_lowerCAmelCase			,     truncation=_lowerCAmelCase			,     max_length=10			,     return_tensors="pt"		)
     _lowerCAmelCase				=      targets["input_ids"]
     _lowerCAmelCase				=      shift_tokens_right(_lowerCAmelCase			,     self.tokenizer.pad_token_id		)

     self.assertEqual(batch.input_ids.shape[1]			,     3		)
     self.assertEqual(batch.decoder_input_ids.shape[1]			,     10		)





 @require_torch
 def 							_snake_case   (	self		)    ->							Optional[Any]:
     _lowerCAmelCase				=      self.tokenizer._build_translation_inputs(
         "A test"			,     return_tensors="pt"			,     src_lang="en_XX"			,     tgt_lang="java"		)

     self.assertEqual(
         nested_simplify(_lowerCAmelCase		)			,     {
             # A, test, EOS, en_XX
             "input_ids": [[150, 242, 2, 50003]],
             "attention_mask": [[1, 1, 1, 1]],
             # java
             "forced_bos_token_id": 50001,
         }			,     )









'''simple docstring'''


from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
       from ..models.auto import AutoTokenizer

class    lowerCAmelCase_		:




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							Tuple:
     raise NotImplementedError()





 def 							_snake_case   (	self		)    ->							Optional[int]:
     raise NotImplementedError()







class    lowerCAmelCase_		(		__magic_name__		):




 def __init__(	self			,     _lowerCAmelCase			,     _lowerCAmelCase = False			,     **_lowerCAmelCase		)    ->							str:
     _lowerCAmelCase				=      tokenizer
     _lowerCAmelCase				=      skip_prompt
     _lowerCAmelCase				=      decode_kwargs

     # variables used in the streaming process
     _lowerCAmelCase				=      []
     _lowerCAmelCase				=      0
     _lowerCAmelCase				=      True




 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							List[Any]:
     if len(value.shape		) > 1 and value.shape[0] > 1:
         raise ValueError("TextStreamer only supports batch size 1"		)
     elif len(value.shape		) > 1:
         _lowerCAmelCase				=      value[0]

     if self.skip_prompt and self.next_tokens_are_prompt:
         _lowerCAmelCase				=      False
         return

     # Add the new token to the cache and decodes the entire thing.
     self.token_cache.extend(value.tolist()		)
     _lowerCAmelCase				=      self.tokenizer.decode(self.token_cache			,     **self.decode_kwargs		)

     # After the symbol for a new line, we flush the cache.
     if text.endswith("\n"		):
         _lowerCAmelCase				=      text[self.print_len :]
         _lowerCAmelCase				=      []
         _lowerCAmelCase				=      0
     # If the last token is a CJK character, we print the characters.
     elif len(_lowerCAmelCase		) > 0 and self._is_chinese_char(ord(text[-1]		)		):
         _lowerCAmelCase				=      text[self.print_len :]
         self.print_len += len(_lowerCAmelCase		)
     # Otherwise, prints until the last space char (simple heuristic to avoid printing incomplete words,
     # which may change with the subsequent token -- there are probably smarter ways to do this!)
     else:
         _lowerCAmelCase				=      text[self.print_len : text.rfind(" "		) + 1]
         self.print_len += len(_lowerCAmelCase		)

     self.on_finalized_text(_lowerCAmelCase		)




 def 							_snake_case   (	self		)    ->							Union[str, Any]:
     # Flush the cache, if it exists
     if len(self.token_cache		) > 0:
         _lowerCAmelCase				=      self.tokenizer.decode(self.token_cache			,     **self.decode_kwargs		)
         _lowerCAmelCase				=      text[self.print_len :]
         _lowerCAmelCase				=      []
         _lowerCAmelCase				=      0
     else:
         _lowerCAmelCase				=      ""

     _lowerCAmelCase				=      True
     self.on_finalized_text(_lowerCAmelCase			,     stream_end=_lowerCAmelCase		)




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase = False		)    ->							List[Any]:
     print(_lowerCAmelCase			,     flush=_lowerCAmelCase			,     end="" if not stream_end else None		)





 def 							_snake_case   (	self			,     _lowerCAmelCase		)    ->							Tuple:
     # This defines a "chinese character" as anything in the CJK Unicode block:
     #   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
     #
     # Note that the CJK Unicode block is NOT all Japanese and Korean characters,
     # despite its name. The modern Korean Hangul alphabet is a different block,
     # as is Japanese Hiragana and Katakana. Those alphabets are used to write
     # space-separated words, so they are not treated specially and handled
     # like the all of the other languages.
     if (
         (cp >= 0X4E_00 and cp <= 0X9F_FF)
         or (cp >= 0X34_00 and cp <= 0X4D_BF)  #
         or (cp >= 0X2_00_00 and cp <= 0X2_A6_DF)  #
         or (cp >= 0X2_A7_00 and cp <= 0X2_B7_3F)  #
         or (cp >= 0X2_B7_40 and cp <= 0X2_B8_1F)  #
         or (cp >= 0X2_B8_20 and cp <= 0X2_CE_AF)  #
         or (cp >= 0XF9_00 and cp <= 0XFA_FF)
         or (cp >= 0X2_F8_00 and cp <= 0X2_FA_1F)  #
     ):  #
         return True

     return False







class    lowerCAmelCase_		(		__magic_name__		):




 def __init__(	self			,     _lowerCAmelCase			,     _lowerCAmelCase = False			,     _lowerCAmelCase = None			,     **_lowerCAmelCase		)    ->							Dict:
     super().__init__(_lowerCAmelCase			,     _lowerCAmelCase			,     **_lowerCAmelCase		)
     _lowerCAmelCase				=      Queue()
     _lowerCAmelCase				=      None
     _lowerCAmelCase				=      timeout




 def 							_snake_case   (	self			,     _lowerCAmelCase			,     _lowerCAmelCase = False		)    ->							Any:
     self.text_queue.put(_lowerCAmelCase			,     timeout=self.timeout		)
     if stream_end:
         self.text_queue.put(self.stop_signal			,     timeout=self.timeout		)




 def __iter__(	self		)    ->							int:
     return self





 def 							_snake_case   (	self		)    ->							Tuple:
     _lowerCAmelCase				=      self.text_queue.get(timeout=self.timeout		)
     if value == self.stop_signal:
         raise StopIteration()
     else:
         return value


