


'''simple docstring'''



import json
import os
import re
import unittest

from transformers import CodeGenTokenizer, CodeGenTokenizerFast
from transformers.models.codegen.tokenization_codegen import VOCAB_FILES_NAMES
from transformers.testing_utils import require_tokenizers, slow

from ...test_tokenization_common import TokenizerTesterMixin


@require_tokenizers
class 						_snake_case   (       a__    ,				unittest.TestCase		):
					lowerCAmelCase :int						=   CodeGenTokenizer
					lowerCAmelCase :str						=   CodeGenTokenizerFast
					lowerCAmelCase :List[Any]						=   True
					lowerCAmelCase :Union[str, Any]						=   {'''add_prefix_space''': True}
					lowerCAmelCase :Optional[int]						=   False




					def 			snake_case__  ( self):
								super().setUp()

								# Adapted from Sennrich et al. 2015 and https://github.com/rsennrich/subword-nmt
								UpperCAmelCase__					:   Optional[Any]       				=       [
								    """l""",
								    """o""",
								    """w""",
								    """e""",
								    """r""",
								    """s""",
								    """t""",
								    """i""",
								    """d""",
								    """n""",
								    """\u0120""",
								    """\u0120l""",
								    """\u0120n""",
								    """\u0120lo""",
								    """\u0120low""",
								    """er""",
								    """\u0120lowest""",
								    """\u0120newer""",
								    """\u0120wider""",
								    """<unk>""",
								    """<|endoftext|>""",
								]
								UpperCAmelCase__					:   Tuple       				=       dict(zip(_lowerCamelCase							,							range(len(_lowerCamelCase))))
								UpperCAmelCase__					:   int       				=       ["""#version: 0.2""", """\u0120 l""", """\u0120l o""", """\u0120lo w""", """e r""", """"""]
								UpperCAmelCase__					:   Tuple       				=       {"""unk_token""": """<unk>"""}

								UpperCAmelCase__					:   str       				=       os.path.join(self.tmpdirname							,							VOCAB_FILES_NAMES["""vocab_file"""])
								UpperCAmelCase__					:   List[str]       				=       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(_lowerCamelCase) + """\n""")
								with open(self.merges_file							,							"""w"""							,							encoding="""utf-8""") as fp:
											fp.write("""\n""".join(_lowerCamelCase))




					def 			snake_case__  ( self							,							**_lowerCamelCase):
								kwargs.update(self.special_tokens_map)
								return CodeGenTokenizer.from_pretrained(self.tmpdirname							,							**_lowerCamelCase)




					def 			snake_case__  ( self							,							**_lowerCamelCase):
								kwargs.update(self.special_tokens_map)
								return CodeGenTokenizerFast.from_pretrained(self.tmpdirname							,							**_lowerCamelCase)




					def 			snake_case__  ( self							,							_lowerCamelCase):
								UpperCAmelCase__					:   List[str]       				=       """lower newer"""
								UpperCAmelCase__					:   List[str]       				=       """lower newer"""
								return input_text, output_text




					def 			snake_case__  ( self):
								UpperCAmelCase__					:   Optional[int]       				=       CodeGenTokenizer(self.vocab_file							,							self.merges_file							,							**self.special_tokens_map)
								UpperCAmelCase__					:   str       				=       """lower newer"""
								UpperCAmelCase__					:   List[Any]       				=       ["""\u0120low""", """er""", """\u0120""", """n""", """e""", """w""", """er"""]
								UpperCAmelCase__					:   Any       				=       tokenizer.tokenize(_lowerCamelCase							,							add_prefix_space=_lowerCamelCase)
								self.assertListEqual(_lowerCamelCase							,							_lowerCamelCase)

								UpperCAmelCase__					:   Union[str, Any]       				=       tokens + [tokenizer.unk_token]
								UpperCAmelCase__					:   str       				=       [14, 15, 10, 9, 3, 2, 15, 19]
								self.assertListEqual(tokenizer.convert_tokens_to_ids(_lowerCamelCase)							,							_lowerCamelCase)




					def 			snake_case__  ( self):
								if not self.test_rust_tokenizer:
											return

								UpperCAmelCase__					:   Optional[Any]       				=       self.get_tokenizer()
								UpperCAmelCase__					:   str       				=       self.get_rust_tokenizer(add_prefix_space=_lowerCamelCase)

								UpperCAmelCase__					:   Union[str, Any]       				=       """lower newer"""

								# Testing tokenization
								UpperCAmelCase__					:   Dict       				=       tokenizer.tokenize(_lowerCamelCase							,							add_prefix_space=_lowerCamelCase)
								UpperCAmelCase__					:   Union[str, Any]       				=       rust_tokenizer.tokenize(_lowerCamelCase)
								self.assertListEqual(_lowerCamelCase							,							_lowerCamelCase)

								# Testing conversion to ids without special tokens
								UpperCAmelCase__					:   List[str]       				=       tokenizer.encode(_lowerCamelCase							,							add_special_tokens=_lowerCamelCase							,							add_prefix_space=_lowerCamelCase)
								UpperCAmelCase__					:   int       				=       rust_tokenizer.encode(_lowerCamelCase							,							add_special_tokens=_lowerCamelCase)
								self.assertListEqual(_lowerCamelCase							,							_lowerCamelCase)

								# Testing conversion to ids with special tokens
								UpperCAmelCase__					:   List[Any]       				=       self.get_rust_tokenizer(add_prefix_space=_lowerCamelCase)
								UpperCAmelCase__					:   Any       				=       tokenizer.encode(_lowerCamelCase							,							add_prefix_space=_lowerCamelCase)
								UpperCAmelCase__					:   Union[str, Any]       				=       rust_tokenizer.encode(_lowerCamelCase)
								self.assertListEqual(_lowerCamelCase							,							_lowerCamelCase)

								# Testing the unknown token
								UpperCAmelCase__					:   int       				=       tokens + [rust_tokenizer.unk_token]
								UpperCAmelCase__					:   Tuple       				=       [14, 15, 10, 9, 3, 2, 15, 19]
								self.assertListEqual(rust_tokenizer.convert_tokens_to_ids(_lowerCamelCase)							,							_lowerCamelCase)




					def 			snake_case__  ( self							,							*_lowerCamelCase							,							**_lowerCamelCase):
								# It's very difficult to mix/test pretokenization with byte-level
								# And get both CodeGen and Roberta to work at the same time (mostly an issue of adding a space before the string)
								pass




					def 			snake_case__  ( self							,							_lowerCamelCase=15):
								for tokenizer, pretrained_name, kwargs in self.tokenizers_list:
											with self.subTest(f'''{tokenizer.__class__.__name__} ({pretrained_name})'''):
														UpperCAmelCase__					:   int       				=       self.rust_tokenizer_class.from_pretrained(_lowerCamelCase							,							**_lowerCamelCase)

														# Simple input
														UpperCAmelCase__					:   Any       				=       """This is a simple input"""
														UpperCAmelCase__					:   Optional[Any]       				=       ["""This is a simple input 1""", """This is a simple input 2"""]
														UpperCAmelCase__					:   Dict       				=       ("""This is a simple input""", """This is a pair""")
														UpperCAmelCase__					:   str       				=       [
														    ("""This is a simple input 1""", """This is a simple input 2"""),
														    ("""This is a simple pair 1""", """This is a simple pair 2"""),
														]

														# Simple input tests
														self.assertRaises(_lowerCamelCase							,							tokenizer_r.encode							,							_lowerCamelCase							,							max_length=_lowerCamelCase							,							padding="""max_length""")

														# Simple input
														self.assertRaises(_lowerCamelCase							,							tokenizer_r.encode_plus							,							_lowerCamelCase							,							max_length=_lowerCamelCase							,							padding="""max_length""")

														# Simple input
														self.assertRaises(
														    _lowerCamelCase							,							tokenizer_r.batch_encode_plus							,							_lowerCamelCase							,							max_length=_lowerCamelCase							,							padding="""max_length"""							,							)

														# Pair input
														self.assertRaises(_lowerCamelCase							,							tokenizer_r.encode							,							_lowerCamelCase							,							max_length=_lowerCamelCase							,							padding="""max_length""")

														# Pair input
														self.assertRaises(_lowerCamelCase							,							tokenizer_r.encode_plus							,							_lowerCamelCase							,							max_length=_lowerCamelCase							,							padding="""max_length""")

														# Pair input
														self.assertRaises(
														    _lowerCamelCase							,							tokenizer_r.batch_encode_plus							,							_lowerCamelCase							,							max_length=_lowerCamelCase							,							padding="""max_length"""							,							)




					def 			snake_case__  ( self):
								UpperCAmelCase__					:   List[Any]       				=       CodeGenTokenizer.from_pretrained(self.tmpdirname							,							pad_token="""<pad>""")

								# Simple input
								UpperCAmelCase__					:   int       				=       """This is a simple input"""
								UpperCAmelCase__					:   Tuple       				=       ["""This is a simple input looooooooong""", """This is a simple input"""]
								UpperCAmelCase__					:   int       				=       ("""This is a simple input""", """This is a pair""")
								UpperCAmelCase__					:   int       				=       [
								    ("""This is a simple input loooooong""", """This is a simple input"""),
								    ("""This is a simple pair loooooong""", """This is a simple pair"""),
								]

								UpperCAmelCase__					:   int       				=       tokenizer.pad_token_id

								UpperCAmelCase__					:   List[str]       				=       tokenizer(_lowerCamelCase							,							padding="""max_length"""							,							max_length=30							,							return_tensors="""np""")
								UpperCAmelCase__					:   List[str]       				=       tokenizer(_lowerCamelCase							,							padding=_lowerCamelCase							,							truncate=_lowerCamelCase							,							return_tensors="""np""")
								UpperCAmelCase__					:   Any       				=       tokenizer(*_lowerCamelCase							,							padding="""max_length"""							,							max_length=60							,							return_tensors="""np""")
								UpperCAmelCase__					:   int       				=       tokenizer(_lowerCamelCase							,							padding=_lowerCamelCase							,							truncate=_lowerCamelCase							,							return_tensors="""np""")

								# s
								# test single string max_length padding
								self.assertEqual(out_s["""input_ids"""].shape[-1]							,							30)
								self.assertTrue(pad_token_id in out_s["""input_ids"""])
								self.assertTrue(0 in out_s["""attention_mask"""])

								# s2
								# test automatic padding
								self.assertEqual(out_sa["""input_ids"""].shape[-1]							,							33)
								# long slice doesn't have padding
								self.assertFalse(pad_token_id in out_sa["""input_ids"""][0])
								self.assertFalse(0 in out_sa["""attention_mask"""][0])
								# short slice does have padding
								self.assertTrue(pad_token_id in out_sa["""input_ids"""][1])
								self.assertTrue(0 in out_sa["""attention_mask"""][1])

								# p
								# test single pair max_length padding
								self.assertEqual(out_p["""input_ids"""].shape[-1]							,							60)
								self.assertTrue(pad_token_id in out_p["""input_ids"""])
								self.assertTrue(0 in out_p["""attention_mask"""])

								# p2
								# test automatic padding pair
								self.assertEqual(out_pa["""input_ids"""].shape[-1]							,							52)
								# long slice pair doesn't have padding
								self.assertFalse(pad_token_id in out_pa["""input_ids"""][0])
								self.assertFalse(0 in out_pa["""attention_mask"""][0])
								# short slice pair does have padding
								self.assertTrue(pad_token_id in out_pa["""input_ids"""][1])
								self.assertTrue(0 in out_pa["""attention_mask"""][1])




					def 			snake_case__  ( self):
								UpperCAmelCase__					:   str       				=       """$$$"""
								UpperCAmelCase__					:   Optional[Any]       				=       CodeGenTokenizer.from_pretrained(self.tmpdirname							,							bos_token=_lowerCamelCase							,							add_bos_token=_lowerCamelCase)

								UpperCAmelCase__					:   int       				=       """This is a simple input"""
								UpperCAmelCase__					:   List[Any]       				=       ["""This is a simple input 1""", """This is a simple input 2"""]

								UpperCAmelCase__					:   int       				=       tokenizer.bos_token_id

								UpperCAmelCase__					:   Any       				=       tokenizer(_lowerCamelCase)
								UpperCAmelCase__					:   List[str]       				=       tokenizer(_lowerCamelCase)

								self.assertEqual(out_s.input_ids[0]							,							_lowerCamelCase)
								self.assertTrue(all(o[0] == bos_token_id for o in out_sa.input_ids))

								UpperCAmelCase__					:   Union[str, Any]       				=       tokenizer.decode(out_s.input_ids)
								UpperCAmelCase__					:   str       				=       tokenizer.batch_decode(out_sa.input_ids)

								self.assertEqual(decode_s.split()[0]							,							_lowerCamelCase)
								self.assertTrue(all(d.split()[0] == bos_token for d in decode_sa))




					@slow
					def 			snake_case__  ( self):
								UpperCAmelCase__					:   List[Any]       				=       CodeGenTokenizer.from_pretrained("""Salesforce/codegen-350M-mono""")

								UpperCAmelCase__					:   List[Any]       				=       """\nif len_a > len_b:\n    result = a\nelse:\n    result = b\n\n\n\n#"""
								UpperCAmelCase__					:   Union[str, Any]       				=       """\nif len_a > len_b:      result = a\nelse:      result = b"""

								UpperCAmelCase__					:   str       				=       tokenizer.encode(_lowerCamelCase)
								UpperCAmelCase__					:   Any       				=       ["""^#""", re.escape("""<|endoftext|>"""), """^'''""", """^\"\"\"""", """\n\n\n"""]
								UpperCAmelCase__					:   str       				=       tokenizer.decode(_lowerCamelCase							,							truncate_before_pattern=_lowerCamelCase)
								self.assertEqual(_lowerCamelCase							,							_lowerCamelCase)




					def 			snake_case__  ( self):
								pass



'''simple docstring'''



import os
from typing import Optional

import fsspec
from fsspec.archive import AbstractArchiveFileSystem
from fsspec.utils import DEFAULT_BLOCK_SIZE


class 						_snake_case   (       a__		):
					lowerCAmelCase :Optional[int]						=   ''''''
					lowerCAmelCase :str						=   (
					    None  # protocol passed in prefix to the url. ex: "gzip", for gzip://file.txt::http://foo.bar/file.txt.gz
					)
					lowerCAmelCase :str						=   None  # compression type in fsspec. ex: "gzip"
					lowerCAmelCase :str						=   None  # extension of the filename to strip. ex: "".gz" to get file.txt from file.txt.gz




					def __init__( self							,							_lowerCamelCase = ""							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							**_lowerCamelCase):
								super().__init__(self							,							**_lowerCamelCase)
								# always open as "rb" since fsspec can then use the TextIOWrapper to make it work for "r" mode
								UpperCAmelCase__					:   Optional[Any]       				=       fsspec.open(
								    _lowerCamelCase							,							mode="""rb"""							,							protocol=_lowerCamelCase							,							compression=self.compression							,							client_kwargs={
								        """requote_redirect_url""": False,  # see https://github.com/huggingface/datasets/pull/5459
								        """trust_env""": True,  # Enable reading proxy env variables.
								        **(target_options or {}).pop("""client_kwargs"""							,							{}),  # To avoid issues if it was already passed.
								    }							,							**(target_options or {})							,							)
								UpperCAmelCase__					:   List[Any]       				=       os.path.basename(self.file.path.split("""::""")[0])
								UpperCAmelCase__					:   Dict       				=       (
								    self.compressed_name[: self.compressed_name.rindex(""".""")]
								    if """.""" in self.compressed_name
								    else self.compressed_name
								)
								UpperCAmelCase__					:   Tuple       				=       None




					@classmethod
					def 			snake_case__  ( cls							,							_lowerCamelCase):
								# compressed file paths are always relative to the archive root
								return super()._strip_protocol(_lowerCamelCase).lstrip("""/""")




					def 			snake_case__  ( self):
								if self.dir_cache is None:
											UpperCAmelCase__					:   Optional[Any]       				=       {**self.file.fs.info(self.file.path), """name""": self.uncompressed_name}
											UpperCAmelCase__					:   Union[str, Any]       				=       {f["""name"""]: f}




					def 			snake_case__  ( self							,							_lowerCamelCase):
								return self.file.open().read()




					def 			snake_case__  ( self							,							_lowerCamelCase							,							_lowerCamelCase = "rb"							,							_lowerCamelCase=None							,							_lowerCamelCase=True							,							_lowerCamelCase=None							,							**_lowerCamelCase							,							):
								UpperCAmelCase__					:   List[str]       				=       self._strip_protocol(_lowerCamelCase)
								if mode != "rb":
											raise ValueError(f'''Tried to read with mode {mode} on file {self.file.path} opened with mode \'rb\'''')
								return self.file.open()


class 						_snake_case   (       a__		):
					lowerCAmelCase :Dict						=   '''bz2'''
					lowerCAmelCase :List[str]						=   '''bz2'''
					lowerCAmelCase :Dict						=   '''.bz2'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :int						=   '''gzip'''
					lowerCAmelCase :Tuple						=   '''gzip'''
					lowerCAmelCase :str						=   '''.gz'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :List[str]						=   '''lz4'''
					lowerCAmelCase :Any						=   '''lz4'''
					lowerCAmelCase :int						=   '''.lz4'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :Union[str, Any]						=   '''xz'''
					lowerCAmelCase :int						=   '''xz'''
					lowerCAmelCase :List[Any]						=   '''.xz'''


class 						_snake_case   (       a__		):
					lowerCAmelCase :Tuple						=   '''zstd'''
					lowerCAmelCase :List[str]						=   '''zstd'''
					lowerCAmelCase :Union[str, Any]						=   '''.zst'''




					def __init__( self							,							_lowerCamelCase							,							_lowerCamelCase = "rb"							,							_lowerCamelCase = None							,							_lowerCamelCase = None							,							_lowerCamelCase = DEFAULT_BLOCK_SIZE							,							**_lowerCamelCase							,							):
								super().__init__(
								    fo=_lowerCamelCase							,							mode=_lowerCamelCase							,							target_protocol=_lowerCamelCase							,							target_options=_lowerCamelCase							,							block_size=_lowerCamelCase							,							**_lowerCamelCase							,							)
								# We need to wrap the zstd decompressor to avoid this error in fsspec==2021.7.0 and zstandard==0.15.2:
								#
								# File "/Users/user/.virtualenvs/hf-datasets/lib/python3.7/site-packages/fsspec/core.py", line 145, in open
								#     out.close = close
								# AttributeError: 'zstd.ZstdDecompressionReader' object attribute 'close' is read-only
								#
								# see https://github.com/intake/filesystem_spec/issues/725
								UpperCAmelCase__					:   Dict       				=       self.file.__enter__


								class 						_snake_case   :




													def __init__( self							,							_lowerCamelCase):
																UpperCAmelCase__					:   Optional[int]       				=       file_




													def __enter__( self):
																self._file.__enter__()
																return self




													def __exit__( self							,							*_lowerCamelCase							,							**_lowerCamelCase):
																self._file.__exit__(*_lowerCamelCase							,							**_lowerCamelCase)




													def __iter__( self):
																return iter(self._file)




													def 			snake_case__  ( self):
																return next(self._file)




													def __getattr__( self							,							_lowerCamelCase):
																return getattr(self._file							,							_lowerCamelCase)

								def fixed_enter(*_lowerCamelCase							,							**_lowerCamelCase):
											return WrappedFile(_enter(*_lowerCamelCase							,							**_lowerCamelCase))

								UpperCAmelCase__					:   List[Any]       				=       fixed_enter