


"""simple docstring"""





import json
import os
from functools import lru_cache
from typing import TYPE_CHECKING, List, Optional, Tuple

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...utils import logging


if TYPE_CHECKING:
 from transformers.pipelines.conversational import Conversation

snake_case__		:  List[Any]     					=					logging.get_logger(__name__)

snake_case__		:  int     					=					{
    'vocab_file': 'vocab.json',
    'merges_file': 'merges.txt',
}

snake_case__		:  Tuple     					=					{
    'vocab_file': {
        'gpt2': 'https://huggingface.co/gpt2/resolve/main/vocab.json',
        'gpt2-medium': 'https://huggingface.co/gpt2-medium/resolve/main/vocab.json',
        'gpt2-large': 'https://huggingface.co/gpt2-large/resolve/main/vocab.json',
        'gpt2-xl': 'https://huggingface.co/gpt2-xl/resolve/main/vocab.json',
        'distilgpt2': 'https://huggingface.co/distilgpt2/resolve/main/vocab.json',
    },
    'merges_file': {
        'gpt2': 'https://huggingface.co/gpt2/resolve/main/merges.txt',
        'gpt2-medium': 'https://huggingface.co/gpt2-medium/resolve/main/merges.txt',
        'gpt2-large': 'https://huggingface.co/gpt2-large/resolve/main/merges.txt',
        'gpt2-xl': 'https://huggingface.co/gpt2-xl/resolve/main/merges.txt',
        'distilgpt2': 'https://huggingface.co/distilgpt2/resolve/main/merges.txt',
    },
}

snake_case__		:  List[Any]     					=					{
    'gpt2': 1_024,
    'gpt2-medium': 1_024,
    'gpt2-large': 1_024,
    'gpt2-xl': 1_024,
    'distilgpt2': 1_024,
}


@lru_cache()
def   _snake_case    (    ):
      lowerCAmelCase       :						Optional[int]							= (
          list(range(ord('''!'''      )			, ord('''~'''      ) + 1      )      ) + list(range(ord('''¡'''      )			, ord('''¬'''      ) + 1      )      ) + list(range(ord('''®'''      )			, ord('''ÿ'''      ) + 1      )      )
      )
      lowerCAmelCase       :						Union[str, Any]							= bs[:]
      lowerCAmelCase       :						Optional[int]							= 0
      for b in range(2**8      ):
            if b not in bs:
                  bs.append(lowercase__      )
                  cs.append(2**8 + n      )
                  n += 1
      lowerCAmelCase       :						List[str]							= [chr(lowercase__      ) for n in cs]
      return dict(zip(lowercase__			, lowercase__      )      )


def   _snake_case    (    _snake_case					:      Dict      ):
      lowerCAmelCase       :						str							= set()
      lowerCAmelCase       :						Optional[int]							= word[0]
      for char in word[1:]:
            pairs.add((prev_char, char)      )
            lowerCAmelCase       :						Any							= char
      return pairs






class 		snake_case_(							__UpperCAmelCase      ):
     __UpperCamelCase	=       VOCAB_FILES_NAMES
     __UpperCamelCase	=       PRETRAINED_VOCAB_FILES_MAP
     __UpperCamelCase	=       PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     __UpperCamelCase	=       ['''input_ids''', '''attention_mask''']






     def __init__(						self	:       Optional[int]       ,					UpperCamelCase_	:       int       ,					UpperCamelCase_	:       Optional[Any]       ,					UpperCamelCase_	:       Dict="replace"       ,					UpperCamelCase_	:       Optional[Any]="<|endoftext|>"       ,					UpperCamelCase_	:       Any="<|endoftext|>"       ,					UpperCamelCase_	:       Union[str, Any]="<|endoftext|>"       ,					UpperCamelCase_	:       int=None       ,					UpperCamelCase_	:       str=False       ,					UpperCamelCase_	:       str=False       ,					**UpperCamelCase_	:       str       ,					):
           lowerCAmelCase       :						List[str]							= AddedToken(__snake_case       ,					lstrip=__snake_case       ,					rstrip=__snake_case					) if isinstance(__snake_case       ,					__snake_case					) else bos_token
           lowerCAmelCase       :						Union[str, Any]							= AddedToken(__snake_case       ,					lstrip=__snake_case       ,					rstrip=__snake_case					) if isinstance(__snake_case       ,					__snake_case					) else eos_token
           lowerCAmelCase       :						Union[str, Any]							= AddedToken(__snake_case       ,					lstrip=__snake_case       ,					rstrip=__snake_case					) if isinstance(__snake_case       ,					__snake_case					) else unk_token
           lowerCAmelCase       :						str							= AddedToken(__snake_case       ,					lstrip=__snake_case       ,					rstrip=__snake_case					) if isinstance(__snake_case       ,					__snake_case					) else pad_token
           super().__init__(
               errors=__snake_case       ,					unk_token=__snake_case       ,					bos_token=__snake_case       ,					eos_token=__snake_case       ,					pad_token=__snake_case       ,					add_prefix_space=__snake_case       ,					add_bos_token=__snake_case       ,					**__snake_case       ,					)
           lowerCAmelCase       :						Optional[int]							= add_bos_token

           with open(__snake_case       ,					encoding='''utf-8'''					) as vocab_handle:
                 lowerCAmelCase       :						Any							= json.load(__snake_case					)
           lowerCAmelCase       :						int							= {v: k for k, v in self.encoder.items()}
           lowerCAmelCase       :						Tuple							= errors  # how to handle errors in decoding
           lowerCAmelCase       :						Union[str, Any]							= bytes_to_unicode()
           lowerCAmelCase       :						Any							= {v: k for k, v in self.byte_encoder.items()}
           with open(__snake_case       ,					encoding='''utf-8'''					) as merges_handle:
                 lowerCAmelCase       :						List[Any]							= merges_handle.read().split('''\n'''					)[1:-1]
           lowerCAmelCase       :						Dict							= [tuple(merge.split()					) for merge in bpe_merges]
           lowerCAmelCase       :						Any							= dict(zip(__snake_case       ,					range(len(__snake_case					)					)					)					)
           lowerCAmelCase       :						List[str]							= {}
           lowerCAmelCase       :						Dict							= add_prefix_space

           # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
           lowerCAmelCase       :						Tuple							= re.compile(r'''\'s|\'t|\'re|\'ve|\'m|\'ll|\'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+'''					)






     @property
     def   lowerCamelCase__  (						self	:       str					):
           return len(self.encoder					)






     def   lowerCamelCase__  (						self	:       Optional[int]					):
           return dict(self.encoder       ,					**self.added_tokens_encoder					)






     def   lowerCamelCase__  (						self	:       int       ,					UpperCamelCase_	:       Dict					):





           if token in self.cache:
                 return self.cache[token]
           lowerCAmelCase       :						Union[str, Any]							= tuple(__snake_case					)
           lowerCAmelCase       :						Any							= get_pairs(__snake_case					)

           if not pairs:
                 return token

           while True:
                 lowerCAmelCase       :						Union[str, Any]							= min(__snake_case       ,					key=lambda UpperCamelCase_					: self.bpe_ranks.get(__snake_case       ,					float('''inf'''					)					)					)
                 if bigram not in self.bpe_ranks:
                       break
                 lowerCAmelCase       :						Optional[int]							= bigram
                 lowerCAmelCase       :						Tuple							= []
                 lowerCAmelCase       :						int							= 0
                 while i < len(__snake_case					):
                       try:
                             lowerCAmelCase       :						Any							= word.index(__snake_case       ,					__snake_case					)
                       except ValueError:
                             new_word.extend(word[i:]					)
                             break
                       else:
                             new_word.extend(word[i:j]					)
                             lowerCAmelCase       :						Any							= j

                       if word[i] == first and i < len(__snake_case					) - 1 and word[i + 1] == second:
                             new_word.append(first + second					)
                             i += 2
                       else:
                             new_word.append(word[i]					)
                             i += 1
                 lowerCAmelCase       :						int							= tuple(__snake_case					)
                 lowerCAmelCase       :						str							= new_word
                 if len(__snake_case					) == 1:
                       break
                 else:
                       lowerCAmelCase       :						int							= get_pairs(__snake_case					)
           lowerCAmelCase       :						Any							= ' '.join(__snake_case					)
           lowerCAmelCase       :						Tuple							= word
           return word






     def   lowerCamelCase__  (						self	:       Union[str, Any]       ,					UpperCamelCase_	:       Optional[int]       ,					UpperCamelCase_	:       Optional[Any]=None					):





           if self.add_bos_token:
                 lowerCAmelCase       :						Dict							= [self.bos_token_id]
           else:
                 lowerCAmelCase       :						Optional[Any]							= []

           lowerCAmelCase       :						Dict							= bos_token_ids + token_ids_a

           if token_ids_a is None:
                 return output

           return output + bos_token_ids + token_ids_a






     def   lowerCamelCase__  (						self	:       Optional[int]       ,					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=__snake_case       ,					token_ids_a=__snake_case       ,					already_has_special_tokens=__snake_case					)

           if not self.add_bos_token:
                 return super().get_special_tokens_mask(
                     token_ids_a=__snake_case       ,					token_ids_a=__snake_case       ,					already_has_special_tokens=__snake_case					)

           if token_ids_a is None:
                 return [1] + ([0] * len(__snake_case					))
           return [1] + ([0] * len(__snake_case					)) + [1] + ([0] * len(__snake_case					))






     def   lowerCamelCase__  (						self	:       Any       ,					UpperCamelCase_	:       Tuple					):
           lowerCAmelCase       :						Any							= []
           for token in re.findall(self.pat       ,					__snake_case					):
                 lowerCAmelCase       :						Any							= ''.join(
                     self.byte_encoder[b] for b in token.encode('''utf-8'''					)					)  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
                 bpe_tokens.extend(bpe_token for bpe_token in self.bpe(__snake_case					).split(''' '''					)					)
           return bpe_tokens






     def   lowerCamelCase__  (						self	:       Tuple       ,					UpperCamelCase_	:       Optional[Any]					):
           return self.encoder.get(__snake_case       ,					self.encoder.get(self.unk_token					)					)






     def   lowerCamelCase__  (						self	:       List[Any]       ,					UpperCamelCase_	:       Union[str, Any]					):
           return self.decoder.get(__snake_case					)






     def   lowerCamelCase__  (						self	:       Optional[Any]       ,					UpperCamelCase_	:       Optional[int]					):
           lowerCAmelCase       :						Any							= ''.join(__snake_case					)
           lowerCAmelCase       :						Any							= bytearray([self.byte_decoder[c] for c in text]					).decode('''utf-8'''       ,					errors=self.errors					)
           return text






     def   lowerCamelCase__  (						self	:       str       ,					UpperCamelCase_	:       str       ,					UpperCamelCase_	:       Optional[str] = None					):





           if not os.path.isdir(__snake_case					):
                 logger.error(F'''Vocabulary path ({save_directory}) should be a directory'''					)
                 return
           lowerCAmelCase       :						List[Any]							= os.path.join(
               __snake_case       ,					(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']					)
           lowerCAmelCase       :						int							= os.path.join(
               __snake_case       ,					(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''merges_file''']					)

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

           lowerCAmelCase       :						Any							= 0
           with open(__snake_case       ,					'''w'''       ,					encoding='''utf-8'''					) as writer:
                 writer.write('''#version: 0.2\n'''					)
                 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!'''					)
                             lowerCAmelCase       :						Union[str, Any]							= token_index
                       writer.write(''' '''.join(__snake_case					) + '''\n'''					)
                       index += 1

           return vocab_file, merge_file






     def   lowerCamelCase__  (						self	:       Tuple       ,					UpperCamelCase_	:       str       ,					UpperCamelCase_	:       Optional[int]=False       ,					**UpperCamelCase_	:       Union[str, Any]					):
           lowerCAmelCase       :						List[Any]							= kwargs.pop('''add_prefix_space'''       ,					self.add_prefix_space					)
           if is_split_into_words or add_prefix_space:
                 lowerCAmelCase       :						Optional[Any]							= ' ' + text
           return (text, kwargs)






     def   lowerCamelCase__  (						self	:       Dict       ,					UpperCamelCase_	:       "Conversation"					):
           lowerCAmelCase       :						str							= []
           for is_user, text in conversation.iter_texts():
                 input_ids.extend(self.encode(__snake_case       ,					add_special_tokens=__snake_case					) + [self.eos_token_id]					)
           if len(__snake_case					) > self.model_max_length:
                 lowerCAmelCase       :						List[Any]							= input_ids[-self.model_max_length :]
           return input_ids









'''simple docstring'''






import csv
import os

import pytest

from datasets import Dataset, DatasetDict, Features, NamedSplit, Value
from datasets.io.csv import CsvDatasetReader, CsvDatasetWriter

from ..utils import assert_arrow_memory_doesnt_increase, assert_arrow_memory_increases





def    lowercase_   (		lowercase__					,	lowercase__    ) ->List[str]:
						assert isinstance(lowercase__					,	lowercase__    )
						assert dataset.num_rows == 4
						assert dataset.num_columns == 3
						assert dataset.column_names == ["col_1", "col_2", "col_3"]
						for feature, expected_dtype in expected_features.items():
												assert dataset.features[feature].dtype == expected_dtype





@pytest.mark.parametrize('keep_in_memory'					,	[False, True]    )
def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->str:
						_snake_case: Optional[int]	       =		tmp_path / 'cache'
						_snake_case: int	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
												_snake_case: Any	       =		CsvDatasetReader(lowercase__					,	cache_dir=lowercase__					,	keep_in_memory=lowercase__    ).read()
						_check_csv_dataset(lowercase__					,	lowercase__    )





@pytest.mark.parametrize(
    'features'					,	[
        None,
        {'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'},
        {'col_1': 'string', 'col_2': 'string', 'col_3': 'string'},
        {'col_1': 'int32', 'col_2': 'int32', 'col_3': 'int32'},
        {'col_1': 'float32', 'col_2': 'float32', 'col_3': 'float32'},
    ]					,	)
def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->List[Any]:
						_snake_case: str	       =		tmp_path / 'cache'
						# CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
						_snake_case: Union[str, Any]	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						_snake_case: List[str]	       =		features.copy() if features else default_expected_features
						_snake_case: str	       =		(
						    Features({feature: Value(lowercase__    ) for feature, dtype in features.items()}    ) if features is not None else None
						)
						_snake_case: Optional[int]	       =		CsvDatasetReader(lowercase__					,	features=lowercase__					,	cache_dir=lowercase__    ).read()
						_check_csv_dataset(lowercase__					,	lowercase__    )





@pytest.mark.parametrize('split'					,	[None, NamedSplit('train'    ), 'train', 'test']    )
def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->List[str]:
						_snake_case: Dict	       =		tmp_path / 'cache'
						_snake_case: int	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						_snake_case: str	       =		CsvDatasetReader(lowercase__					,	cache_dir=lowercase__					,	split=lowercase__    ).read()
						_check_csv_dataset(lowercase__					,	lowercase__    )
						assert dataset.split == split if split else "train"





@pytest.mark.parametrize('path_type'					,	[str, list]    )
def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->List[str]:
						if issubclass(lowercase__					,	lowercase__    ):
												_snake_case: Optional[Any]	       =		csv_path
						elif issubclass(lowercase__					,	lowercase__    ):
												_snake_case: Tuple	       =		[csv_path]
						_snake_case: Tuple	       =		tmp_path / 'cache'
						_snake_case: Any	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						_snake_case: Union[str, Any]	       =		CsvDatasetReader(lowercase__					,	cache_dir=lowercase__    ).read()
						_check_csv_dataset(lowercase__					,	lowercase__    )





def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__=("train",)    ) ->str:
						assert isinstance(lowercase__					,	lowercase__    )
						for split in splits:
												_snake_case: Optional[int]	       =		dataset_dict[split]
												assert dataset.num_rows == 4
												assert dataset.num_columns == 3
												assert dataset.column_names == ["col_1", "col_2", "col_3"]
												for feature, expected_dtype in expected_features.items():
																		assert dataset.features[feature].dtype == expected_dtype





@pytest.mark.parametrize('keep_in_memory'					,	[False, True]    )
def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->Optional[int]:
						_snake_case: Optional[Any]	       =		tmp_path / 'cache'
						_snake_case: Optional[Any]	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						with assert_arrow_memory_increases() if keep_in_memory else assert_arrow_memory_doesnt_increase():
												_snake_case: Dict	       =		CsvDatasetReader({'train': csv_path}					,	cache_dir=lowercase__					,	keep_in_memory=lowercase__    ).read()
						_check_csv_datasetdict(lowercase__					,	lowercase__    )





@pytest.mark.parametrize(
    'features'					,	[
        None,
        {'col_1': 'string', 'col_2': 'int64', 'col_3': 'float64'},
        {'col_1': 'string', 'col_2': 'string', 'col_3': 'string'},
        {'col_1': 'int32', 'col_2': 'int32', 'col_3': 'int32'},
        {'col_1': 'float32', 'col_2': 'float32', 'col_3': 'float32'},
    ]					,	)
def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->Optional[int]:
						_snake_case: List[str]	       =		tmp_path / 'cache'
						# CSV file loses col_1 string dtype information: default now is "int64" instead of "string"
						_snake_case: str	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						_snake_case: str	       =		features.copy() if features else default_expected_features
						_snake_case: int	       =		(
						    Features({feature: Value(lowercase__    ) for feature, dtype in features.items()}    ) if features is not None else None
						)
						_snake_case: Tuple	       =		CsvDatasetReader({'train': csv_path}					,	features=lowercase__					,	cache_dir=lowercase__    ).read()
						_check_csv_datasetdict(lowercase__					,	lowercase__    )





@pytest.mark.parametrize('split'					,	[None, NamedSplit('train'    ), 'train', 'test']    )
def    lowercase_   (		lowercase__					,	lowercase__					,	lowercase__    ) ->List[Any]:
						if split:
												_snake_case: Optional[Any]	       =		{split: csv_path}
						else:
												_snake_case: Tuple	       =		{'train': csv_path, 'test': csv_path}
						_snake_case: Union[str, Any]	       =		tmp_path / 'cache'
						_snake_case: Dict	       =		{'col_1': 'int64', 'col_2': 'int64', 'col_3': 'float64'}
						_snake_case: Any	       =		CsvDatasetReader(lowercase__					,	cache_dir=lowercase__    ).read()
						_check_csv_datasetdict(lowercase__					,	lowercase__					,	splits=list(path.keys()    )    )
						assert all(dataset[split].split == split for split in path.keys()    )





def    lowercase_   (		lowercase__    ) ->Union[str, Any]:
						with open(lowercase__					,	encoding='utf-8'    ) as csvfile:
												yield from csv.reader(lowercase__    )





def    lowercase_   (		lowercase__					,	lowercase__    ) ->str:
						_snake_case: List[Any]	       =		tmp_path / 'cache'
						_snake_case: Dict	       =		os.path.join(lowercase__					,	'tmp.csv'    )
						_snake_case: Optional[Any]	       =		CsvDatasetReader({'train': csv_path}					,	cache_dir=lowercase__    ).read()
						CsvDatasetWriter(dataset['train']					,	lowercase__					,	num_proc=1    ).write()

						_snake_case: Optional[int]	       =		iter_csv_file(lowercase__    )
						_snake_case: List[Any]	       =		iter_csv_file(lowercase__    )

						for rowa, rowa in zip(lowercase__					,	lowercase__    ):
												assert rowa == rowa





def    lowercase_   (		lowercase__					,	lowercase__    ) ->str:
						_snake_case: List[str]	       =		tmp_path / 'cache'
						_snake_case: List[str]	       =		os.path.join(lowercase__					,	'tmp.csv'    )
						_snake_case: List[str]	       =		CsvDatasetReader({'train': csv_path}					,	cache_dir=lowercase__    ).read()
						CsvDatasetWriter(dataset['train']					,	lowercase__					,	num_proc=2    ).write()

						_snake_case: Optional[Any]	       =		iter_csv_file(lowercase__    )
						_snake_case: Tuple	       =		iter_csv_file(lowercase__    )

						for rowa, rowa in zip(lowercase__					,	lowercase__    ):
												assert rowa == rowa





def    lowercase_   (		lowercase__					,	lowercase__    ) ->Any:
						_snake_case: Optional[int]	       =		tmp_path / 'cache'
						_snake_case: Any	       =		os.path.join(lowercase__					,	'tmp.csv'    )
						_snake_case: str	       =		CsvDatasetReader({'train': csv_path}					,	cache_dir=lowercase__    ).read()
						with pytest.raises(lowercase__    ):
												CsvDatasetWriter(dataset['train']					,	lowercase__					,	num_proc=0    )





