





import json
import os
import shutil
import tempfile
import unittest
from multiprocessing import get_context
from pathlib import Path

import datasets
import numpy as np
from datasets import load_dataset
from parameterized import parameterized

from transformers import AutoProcessor
from transformers.models.wavaveca import WavaVecaCTCTokenizer, WavaVecaFeatureExtractor
from transformers.models.wavaveca.tokenization_wavaveca import VOCAB_FILES_NAMES
from transformers.testing_utils import require_pyctcdecode, require_torch, require_torchaudio, slow
from transformers.utils import FEATURE_EXTRACTOR_NAME, is_pyctcdecode_available, is_torch_available

from ..wavaveca.test_feature_extraction_wavaveca import floats_list


if is_pyctcdecode_available():
  from huggingface_hub import snapshot_download
  from pyctcdecode import BeamSearchDecoderCTC

  from transformers.models.wavaveca_with_lm import WavaVecaProcessorWithLM
  from transformers.models.wavaveca_with_lm.processing_wavaveca_with_lm import WavaVecaDecoderWithLMOutput

if is_torch_available():
  from transformers import WavaVecaForCTC





@require_pyctcdecode
class    lowerCAmelCase				(  unittest.TestCase   ):







  def     UpperCAmelCase	(     self    :Optional[Any]   ):






     '''simple docstring'''





     lowercase__					  =    "| <pad> <unk> <s> </s> a b c d e f g h i j k".split()
     lowercase__					  =    dict(zip(_a       ,				range(len(_a   )   )   )   )

     lowercase__					  =    {
         "unk_token": "<unk>",
         "bos_token": "<s>",
         "eos_token": "</s>",
     }
     lowercase__					  =    {
         "feature_size": 1,
         "padding_value": 0.0,
         "sampling_rate": 1_60_00,
         "return_attention_mask": False,
         "do_normalize": True,
     }

     lowercase__					  =    tempfile.mkdtemp()
     lowercase__					  =    os.path.join(self.tmpdirname       ,				VOCAB_FILES_NAMES["vocab_file"]   )
     lowercase__					  =    os.path.join(self.tmpdirname       ,				_a   )
     with open(self.vocab_file       ,				"w"       ,				encoding="utf-8"   ) as fp:
        fp.write(json.dumps(_a   ) + "\n"   )

     with open(self.feature_extraction_file       ,				"w"       ,				encoding="utf-8"   ) as fp:
        fp.write(json.dumps(_a   ) + "\n"   )

     # load decoder from hub
     lowercase__					  =    "hf-internal-testing/ngram-beam-search-decoder"







  def     UpperCAmelCase	(     self    :Any       ,				**_lowercase    :Optional[Any]   ):






     '''simple docstring'''





     lowercase__					  =    self.add_kwargs_tokens_map.copy()
     kwargs.update(_a   )
     return WavaVecaCTCTokenizer.from_pretrained(self.tmpdirname       ,				**_a   )







  def     UpperCAmelCase	(     self    :Union[str, Any]       ,				**_lowercase    :Optional[Any]   ):






     '''simple docstring'''





     return WavaVecaFeatureExtractor.from_pretrained(self.tmpdirname       ,				**_a   )







  def     UpperCAmelCase	(     self    :Union[str, Any]       ,				**_lowercase    :Any   ):






     '''simple docstring'''





     return BeamSearchDecoderCTC.load_from_hf_hub(self.decoder_name       ,				**_a   )







  def     UpperCAmelCase	(     self    :Union[str, Any]   ):






     '''simple docstring'''





     shutil.rmtree(self.tmpdirname   )







  def     UpperCAmelCase	(     self    :Tuple   ):






     '''simple docstring'''





     lowercase__					  =    self.get_tokenizer()
     lowercase__					  =    self.get_feature_extractor()
     lowercase__					  =    self.get_decoder()

     lowercase__					  =    WavaVecaProcessorWithLM(tokenizer=_a       ,				feature_extractor=_a       ,				decoder=_a   )

     processor.save_pretrained(self.tmpdirname   )
     lowercase__					  =    WavaVecaProcessorWithLM.from_pretrained(self.tmpdirname   )

     # tokenizer
     self.assertEqual(processor.tokenizer.get_vocab()       ,				tokenizer.get_vocab()   )
     self.assertIsInstance(processor.tokenizer       ,				_a   )

     # feature extractor
     self.assertEqual(processor.feature_extractor.to_json_string()       ,				feature_extractor.to_json_string()   )
     self.assertIsInstance(processor.feature_extractor       ,				_a   )

     # decoder
     self.assertEqual(processor.decoder._alphabet.labels       ,				decoder._alphabet.labels   )
     self.assertEqual(
         processor.decoder.model_container[decoder._model_key]._unigram_set       ,				decoder.model_container[decoder._model_key]._unigram_set       ,				)
     self.assertIsInstance(processor.decoder       ,				_a   )







  def     UpperCAmelCase	(     self    :Optional[int]   ):






     '''simple docstring'''





     lowercase__					  =    WavaVecaProcessorWithLM(
         tokenizer=self.get_tokenizer()       ,				feature_extractor=self.get_feature_extractor()       ,				decoder=self.get_decoder()   )
     processor.save_pretrained(self.tmpdirname   )

     # make sure that error is thrown when decoder alphabet doesn't match
     lowercase__					  =    WavaVecaProcessorWithLM.from_pretrained(
         self.tmpdirname       ,				alpha=5.0       ,				beta=3.0       ,				score_boundary=-7.0       ,				unk_score_offset=3   )

     # decoder
     self.assertEqual(processor.language_model.alpha       ,				5.0   )
     self.assertEqual(processor.language_model.beta       ,				3.0   )
     self.assertEqual(processor.language_model.score_boundary       ,				-7.0   )
     self.assertEqual(processor.language_model.unk_score_offset       ,				3   )







  def     UpperCAmelCase	(     self    :List[Any]   ):






     '''simple docstring'''





     lowercase__					  =    self.get_tokenizer()
     # add token to trigger raise
     tokenizer.add_tokens(["xx"]   )
     with self.assertRaisesRegex(_a       ,				"include"   ):
        WavaVecaProcessorWithLM(
            tokenizer=_a       ,				feature_extractor=self.get_feature_extractor()       ,				decoder=self.get_decoder()   )







  def     UpperCAmelCase	(     self    :int   ):






     '''simple docstring'''





     lowercase__					  =    self.get_feature_extractor()
     lowercase__					  =    self.get_tokenizer()
     lowercase__					  =    self.get_decoder()

     lowercase__					  =    WavaVecaProcessorWithLM(tokenizer=_a       ,				feature_extractor=_a       ,				decoder=_a   )

     lowercase__					  =    floats_list((3, 10_00)   )

     lowercase__					  =    feature_extractor(_a       ,				return_tensors="np"   )
     lowercase__					  =    processor(_a       ,				return_tensors="np"   )

     for key in input_feat_extract.keys():
        self.assertAlmostEqual(input_feat_extract[key].sum()       ,				input_processor[key].sum()       ,				delta=1e-2   )







  def     UpperCAmelCase	(     self    :List[Any]   ):






     '''simple docstring'''





     lowercase__					  =    self.get_feature_extractor()
     lowercase__					  =    self.get_tokenizer()
     lowercase__					  =    self.get_decoder()

     lowercase__					  =    WavaVecaProcessorWithLM(tokenizer=_a       ,				feature_extractor=_a       ,				decoder=_a   )

     lowercase__					  =    "This is a test string"

     lowercase__					  =    processor(text=_a   )

     lowercase__					  =    tokenizer(_a   )

     for key in encoded_tok.keys():
        self.assertListEqual(encoded_tok[key]       ,				encoded_processor[key]   )







  def     UpperCAmelCase	(     self    :Optional[int]       ,				_lowercase    :List[str]=(2, 10, 16)       ,				_lowercase    :Dict=77   ):






     '''simple docstring'''





     np.random.seed(_a   )
     return np.random.rand(*_a   )







  def     UpperCAmelCase	(     self    :List[Any]   ):






     '''simple docstring'''





     lowercase__					  =    self.get_feature_extractor()
     lowercase__					  =    self.get_tokenizer()
     lowercase__					  =    self.get_decoder()

     lowercase__					  =    WavaVecaProcessorWithLM(tokenizer=_a       ,				feature_extractor=_a       ,				decoder=_a   )

     lowercase__					  =    self._get_dummy_logits(shape=(10, 16)       ,				seed=13   )

     lowercase__					  =    processor.decode(_a   )

     lowercase__					  =    decoder.decode_beams(_a   )[0]

     self.assertEqual(decoded_decoder[0]       ,				decoded_processor.text   )
     self.assertEqual("</s> <s> </s>"       ,				decoded_processor.text   )
     self.assertEqual(decoded_decoder[-2]       ,				decoded_processor.logit_score   )
     self.assertEqual(decoded_decoder[-1]       ,				decoded_processor.lm_score   )







  @parameterized.expand([[None], ["fork"], ["spawn"]]   )
  def     UpperCAmelCase	(     self    :Tuple       ,				_lowercase    :Tuple   ):






     '''simple docstring'''





     lowercase__					  =    self.get_feature_extractor()
     lowercase__					  =    self.get_tokenizer()
     lowercase__					  =    self.get_decoder()

     lowercase__					  =    WavaVecaProcessorWithLM(tokenizer=_a       ,				feature_extractor=_a       ,				decoder=_a   )

     lowercase__					  =    self._get_dummy_logits()

     # note: pool should be instantiated *after* Wav2Vec2ProcessorWithLM.
     #       otherwise, the LM won't be available to the pool's sub-processes.
     # manual logic used to allow parameterized test for both pool=None and pool=Pool(...)
     if pool_context is None:
        lowercase__					  =    processor.batch_decode(_a   )
     else:
        with get_context(_a   ).Pool() as pool:
           lowercase__					  =    processor.batch_decode(_a       ,				_a   )

     lowercase__					  =    list(_a   )

     with get_context("fork"   ).Pool() as p:
        lowercase__					  =    decoder.decode_beams_batch(_a       ,				_a   )

     lowercase__  ,				lowercase__  ,				lowercase__					  =    [], [], []
     for beams in decoded_beams:
        texts_decoder.append(beams[0][0]   )
        logit_scores_decoder.append(beams[0][-2]   )
        lm_scores_decoder.append(beams[0][-1]   )

     self.assertListEqual(_a       ,				decoded_processor.text   )
     self.assertListEqual(["<s> <s> </s>", "<s> <s> <s>"]       ,				decoded_processor.text   )
     self.assertListEqual(_a       ,				decoded_processor.logit_score   )
     self.assertListEqual(_a       ,				decoded_processor.lm_score   )







  def     UpperCAmelCase	(     self    :Tuple   ):






     '''simple docstring'''





     lowercase__					  =    self.get_feature_extractor()
     lowercase__					  =    self.get_tokenizer()
     lowercase__					  =    self.get_decoder()

     lowercase__					  =    WavaVecaProcessorWithLM(tokenizer=_a       ,				feature_extractor=_a       ,				decoder=_a   )

     lowercase__					  =    self._get_dummy_logits()

     lowercase__					  =    15
     lowercase__					  =    -20.0
     lowercase__					  =    -4.0

     lowercase__					  =    processor.batch_decode(
         _a       ,				beam_width=_a       ,				beam_prune_logp=_a       ,				token_min_logp=_a       ,				)
     lowercase__					  =    decoded_processor_out.text

     lowercase__					  =    list(_a   )

     with get_context("fork"   ).Pool() as pool:
        lowercase__					  =    decoder.decode_beams_batch(
            _a       ,				_a       ,				beam_width=_a       ,				beam_prune_logp=_a       ,				token_min_logp=_a       ,				)

     lowercase__					  =    [d[0][0] for d in decoded_decoder_out]
     lowercase__					  =    [d[0][2] for d in decoded_decoder_out]
     lowercase__					  =    [d[0][3] for d in decoded_decoder_out]

     self.assertListEqual(_a       ,				_a   )
     self.assertListEqual(["</s> <s> <s>", "<s> <s> <s>"]       ,				_a   )

     self.assertTrue(np.array_equal(_a       ,				decoded_processor_out.logit_score   )   )
     self.assertTrue(np.allclose([-20.054, -18.447]       ,				_a       ,				atol=1e-3   )   )

     self.assertTrue(np.array_equal(_a       ,				decoded_processor_out.lm_score   )   )
     self.assertTrue(np.allclose([-15.554, -13.9474]       ,				_a       ,				atol=1e-3   )   )







  def     UpperCAmelCase	(     self    :Union[str, Any]   ):






     '''simple docstring'''





     lowercase__					  =    self.get_feature_extractor()
     lowercase__					  =    self.get_tokenizer()
     lowercase__					  =    self.get_decoder()

     lowercase__					  =    WavaVecaProcessorWithLM(tokenizer=_a       ,				feature_extractor=_a       ,				decoder=_a   )

     lowercase__					  =    self._get_dummy_logits()

     lowercase__					  =    2.0
     lowercase__					  =    5.0
     lowercase__					  =    -20.0
     lowercase__					  =    True

     lowercase__					  =    processor.batch_decode(
         _a       ,				alpha=_a       ,				beta=_a       ,				unk_score_offset=_a       ,				lm_score_boundary=_a       ,				)
     lowercase__					  =    decoded_processor_out.text

     lowercase__					  =    list(_a   )
     decoder.reset_params(
         alpha=_a       ,				beta=_a       ,				unk_score_offset=_a       ,				lm_score_boundary=_a       ,				)

     with get_context("fork"   ).Pool() as pool:
        lowercase__					  =    decoder.decode_beams_batch(
            _a       ,				_a       ,				)

     lowercase__					  =    [d[0][0] for d in decoded_decoder_out]

     self.assertListEqual(_a       ,				_a   )
     self.assertListEqual(["<s> </s> <s> </s> </s>", "</s> </s> <s> </s> </s>"]       ,				_a   )
     lowercase__					  =    processor.decoder.model_container[processor.decoder._model_key]
     self.assertEqual(lm_model.alpha       ,				2.0   )
     self.assertEqual(lm_model.beta       ,				5.0   )
     self.assertEqual(lm_model.unk_score_offset       ,				-20.0   )
     self.assertEqual(lm_model.score_boundary       ,				_a   )







  def     UpperCAmelCase	(     self    :Dict   ):






     '''simple docstring'''





     lowercase__					  =    WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"   )

     lowercase__					  =    processor.decoder.model_container[processor.decoder._model_key]
     lowercase__					  =    Path(language_model._kenlm_model.path.decode("utf-8"   )   ).parent.parent.absolute()

     lowercase__					  =    os.listdir(_a   )
     lowercase__					  =    ["alphabet.json", "language_model"]

     downloaded_decoder_files.sort()
     expected_decoder_files.sort()

     # test that only decoder relevant files from
     # https://huggingface.co/hf-internal-testing/processor_with_lm/tree/main
     # are downloaded and none of the rest (e.g. README.md, ...)
     self.assertListEqual(_a       ,				_a   )







  def     UpperCAmelCase	(     self    :List[str]   ):






     '''simple docstring'''





     lowercase__					  =    snapshot_download("hf-internal-testing/processor_with_lm"   )

     lowercase__					  =    WavaVecaProcessorWithLM.from_pretrained(_a   )

     lowercase__					  =    processor.decoder.model_container[processor.decoder._model_key]
     lowercase__					  =    Path(language_model._kenlm_model.path.decode("utf-8"   )   ).parent.parent.absolute()

     lowercase__					  =    os.listdir(_a   )
     lowercase__					  =    os.listdir(_a   )

     local_decoder_files.sort()
     expected_decoder_files.sort()

     # test that both decoder form hub and local files in cache are the same
     self.assertListEqual(_a       ,				_a   )







  def     UpperCAmelCase	(     self    :Any   ):






     '''simple docstring'''





     lowercase__					  =    WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"   )
     lowercase__					  =    AutoProcessor.from_pretrained("hf-internal-testing/processor_with_lm"   )

     lowercase__					  =    floats_list((3, 10_00)   )

     lowercase__					  =    processor_wavaveca(_a       ,				return_tensors="np"   )
     lowercase__					  =    processor_auto(_a       ,				return_tensors="np"   )

     for key in input_wavaveca.keys():
        self.assertAlmostEqual(input_wavaveca[key].sum()       ,				input_auto[key].sum()       ,				delta=1e-2   )

     lowercase__					  =    self._get_dummy_logits()

     lowercase__					  =    processor_wavaveca.batch_decode(_a   )
     lowercase__					  =    processor_auto.batch_decode(_a   )

     self.assertListEqual(decoded_wavaveca.text       ,				decoded_auto.text   )







  def     UpperCAmelCase	(     self    :Any   ):






     '''simple docstring'''





     lowercase__					  =    self.get_feature_extractor()
     lowercase__					  =    self.get_tokenizer()
     lowercase__					  =    self.get_decoder()

     lowercase__					  =    WavaVecaProcessorWithLM(tokenizer=_a       ,				feature_extractor=_a       ,				decoder=_a   )

     self.assertListEqual(
         processor.model_input_names       ,				feature_extractor.model_input_names       ,				msg="`processor` and `feature_extractor` model input names do not match"       ,				)







  @staticmethod
  def     UpperCAmelCase	(     _lowercase    :Union[str, Any]       ,				_lowercase    :Dict   ):






     '''simple docstring'''





     lowercase__					  =    [d[key] for d in offsets]
     return retrieved_list







  def     UpperCAmelCase	(     self    :List[str]   ):






     '''simple docstring'''





     lowercase__					  =    WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"   )
     lowercase__					  =    self._get_dummy_logits()[0]

     lowercase__					  =    processor.decode(_a       ,				output_word_offsets=_a   )
     # check Wav2Vec2CTCTokenizerOutput keys for word
     self.assertEqual(len(outputs.keys()   )       ,				4   )
     self.assertTrue("text" in outputs   )
     self.assertTrue("word_offsets" in outputs   )
     self.assertTrue(isinstance(_a       ,				_a   )   )

     self.assertEqual(" ".join(self.get_from_offsets(outputs["word_offsets"]       ,				"word"   )   )       ,				outputs.text   )
     self.assertListEqual(self.get_from_offsets(outputs["word_offsets"]       ,				"word"   )       ,				["<s>", "<s>", "</s>"]   )
     self.assertListEqual(self.get_from_offsets(outputs["word_offsets"]       ,				"start_offset"   )       ,				[0, 2, 4]   )
     self.assertListEqual(self.get_from_offsets(outputs["word_offsets"]       ,				"end_offset"   )       ,				[1, 3, 5]   )







  def     UpperCAmelCase	(     self    :Optional[Any]   ):






     '''simple docstring'''





     lowercase__					  =    WavaVecaProcessorWithLM.from_pretrained("hf-internal-testing/processor_with_lm"   )
     lowercase__					  =    self._get_dummy_logits()

     lowercase__					  =    processor.batch_decode(_a       ,				output_word_offsets=_a   )

     # check Wav2Vec2CTCTokenizerOutput keys for word
     self.assertEqual(len(outputs.keys()   )       ,				4   )
     self.assertTrue("text" in outputs   )
     self.assertTrue("word_offsets" in outputs   )
     self.assertTrue(isinstance(_a       ,				_a   )   )

     self.assertListEqual(
         [" ".join(self.get_from_offsets(_a       ,				"word"   )   ) for o in outputs["word_offsets"]]       ,				outputs.text   )
     self.assertListEqual(self.get_from_offsets(outputs["word_offsets"][0]       ,				"word"   )       ,				["<s>", "<s>", "</s>"]   )
     self.assertListEqual(self.get_from_offsets(outputs["word_offsets"][0]       ,				"start_offset"   )       ,				[0, 2, 4]   )
     self.assertListEqual(self.get_from_offsets(outputs["word_offsets"][0]       ,				"end_offset"   )       ,				[1, 3, 5]   )







  @slow
  @require_torch
  @require_torchaudio
  def     UpperCAmelCase	(     self    :Any   ):






     '''simple docstring'''





     import torch

     lowercase__					  =    load_dataset("common_voice"       ,				"en"       ,				split="train"       ,				streaming=_a   )
     lowercase__					  =    ds.cast_column("audio"       ,				datasets.Audio(sampling_rate=1_60_00   )   )
     lowercase__					  =    iter(_a   )
     lowercase__					  =    next(_a   )

     lowercase__					  =    AutoProcessor.from_pretrained("patrickvonplaten/wav2vec2-base-100h-with-lm"   )
     lowercase__					  =    WavaVecaForCTC.from_pretrained("patrickvonplaten/wav2vec2-base-100h-with-lm"   )

     # compare to filename `common_voice_en_100038.mp3` of dataset viewer on https://huggingface.co/datasets/common_voice/viewer/en/train
     lowercase__					  =    processor(sample["audio"]["array"]       ,				return_tensors="pt"   ).input_values

     with torch.no_grad():
        lowercase__					  =    model(_a   ).logits.cpu().numpy()

     lowercase__					  =    processor.decode(logits[0]       ,				output_word_offsets=_a   )

     lowercase__					  =    model.config.inputs_to_logits_ratio / processor.feature_extractor.sampling_rate
     lowercase__					  =    [
         {
             "start_time": d["start_offset"] * time_offset,
             "end_time": d["end_offset"] * time_offset,
             "word": d["word"],
         }
         for d in output["word_offsets"]
     ]

     lowercase__					  =    "WHY DOES MILISANDRA LOOK LIKE SHE WANTS TO CONSUME JOHN SNOW ON THE RIVER AT THE WALL"

     # output words
     self.assertEqual(" ".join(self.get_from_offsets(_a       ,				"word"   )   )       ,				_a   )
     self.assertEqual(" ".join(self.get_from_offsets(_a       ,				"word"   )   )       ,				output.text   )

     # output times
     lowercase__					  =    torch.tensor(self.get_from_offsets(_a       ,				"start_time"   )   )
     lowercase__					  =    torch.tensor(self.get_from_offsets(_a       ,				"end_time"   )   )

     # fmt: off
     lowercase__					  =    torch.tensor([1.4199, 1.6599, 2.2599, 3.0, 3.24, 3.5999, 3.7999, 4.0999, 4.26, 4.94, 5.28, 5.6599, 5.78, 5.94, 6.32, 6.5399, 6.6599]   )
     lowercase__					  =    torch.tensor([1.5399, 1.8999, 2.9, 3.16, 3.5399, 3.72, 4.0199, 4.1799, 4.76, 5.1599, 5.5599, 5.6999, 5.86, 6.1999, 6.38, 6.6199, 6.94]   )
     # fmt: on

     self.assertTrue(torch.allclose(_a       ,				_a       ,				atol=0.01   )   )
     self.assertTrue(torch.allclose(_a       ,				_a       ,				atol=0.01   )   )












"""simple docstring"""




import json
import os
from typing import Optional, Tuple

import regex as re

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


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

lowerCAmelCase				:     Optional[int]					=				{
    """vocab_file""": """vocab.json""",
    """merges_file""": """merges.txt""",
}

lowerCAmelCase				:     Optional[Any]					=				{
    """vocab_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-vocab.json"""},
    """merges_file""": {"""ctrl""": """https://raw.githubusercontent.com/salesforce/ctrl/master/ctrl-merges.txt"""},
}

lowerCAmelCase				:     Optional[Any]					=				{
    """ctrl""": 256,
}

lowerCAmelCase				:     int					=				{
    """Pregnancy""": 16_8629,
    """Christianity""": 7675,
    """Explain""": 10_6423,
    """Fitness""": 6_3440,
    """Saving""": 6_3163,
    """Ask""": 2_7171,
    """Ass""": 9_5985,
    """Joke""": 16_3509,
    """Questions""": 4_5622,
    """Thoughts""": 4_9605,
    """Retail""": 5_2342,
    """Feminism""": 16_4338,
    """Writing""": 1_1992,
    """Atheism""": 19_2263,
    """Netflix""": 4_8616,
    """Computing""": 3_9639,
    """Opinion""": 4_3213,
    """Alone""": 4_4967,
    """Funny""": 5_8917,
    """Gaming""": 4_0358,
    """Human""": 4088,
    """India""": 1331,
    """Joker""": 7_7138,
    """Diet""": 3_6206,
    """Legal""": 1_1859,
    """Norman""": 4939,
    """Tip""": 7_2689,
    """Weight""": 5_2343,
    """Movies""": 4_6273,
    """Running""": 2_3425,
    """Science""": 2090,
    """Horror""": 3_7793,
    """Confession""": 6_0572,
    """Finance""": 1_2250,
    """Politics""": 1_6360,
    """Scary""": 19_1985,
    """Support""": 1_2654,
    """Technologies""": 3_2516,
    """Teenage""": 6_6160,
    """Event""": 3_2769,
    """Learned""": 6_7460,
    """Notion""": 18_2770,
    """Wikipedia""": 3_7583,
    """Books""": 6665,
    """Extract""": 7_6050,
    """Confessions""": 10_2701,
    """Conspiracy""": 7_5932,
    """Links""": 6_3674,
    """Narcissus""": 15_0425,
    """Relationship""": 5_4766,
    """Relationships""": 13_4796,
    """Reviews""": 4_1671,
    """News""": 4256,
    """Translation""": 2_6820,
    """multilingual""": 12_8406,
}



def        a__					(							snake_case__							)		->   Optional[Any]:
			lowerCamelCase      =   set()
			lowerCamelCase      =   word[0]
			for char in word[1:]:
						pairs.add((prev_char, char)							)
						lowerCamelCase      =   char

			lowerCamelCase      =   set(snake_case__							)
			return pairs






class        __magic_name__				(			UpperCAmelCase__							):






				'''simple docstring'''
				__UpperCamelCase         =		VOCAB_FILES_NAMES
				__UpperCamelCase         =		PRETRAINED_VOCAB_FILES_MAP
				__UpperCamelCase         =		PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
				__UpperCamelCase         =		CONTROL_CODES





				def __init__(			self						,					_a						,					_a						,					_a="<unk>"						,					**_a					):





							"""simple docstring"""
							super().__init__(unk_token=_a						,					**_a					)

							with open(_a						,					encoding="""utf-8"""					) as vocab_handle:
										lowerCamelCase      =   json.load(_a					)
							lowerCamelCase      =   {v: k for k, v in self.encoder.items()}
							with open(_a						,					encoding="""utf-8"""					) as merges_handle:
										lowerCamelCase      =   merges_handle.read().split("""\n"""					)[1:-1]
							lowerCamelCase      =   [tuple(merge.split()					) for merge in merges]
							lowerCamelCase      =   dict(zip(_a						,					range(len(_a					)					)					)					)
							lowerCamelCase      =   {}





				@property
				def   _lowerCAmelCase							(			self					):





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





				def   _lowerCAmelCase							(			self					):





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





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							if token in self.cache:
										return self.cache[token]
							lowerCamelCase      =   tuple(_a					)
							lowerCamelCase      =   tuple(list(word[:-1]					) + [word[-1] + """</w>"""]					)
							lowerCamelCase      =   get_pairs(_a					)

							if not pairs:
										return token

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

													if word[i] == first and i < len(_a					) - 1 and word[i + 1] == second:
																new_word.append(first + second					)
																i += 2
													else:
																new_word.append(word[i]					)
																i += 1
										lowerCamelCase      =   tuple(_a					)
										lowerCamelCase      =   new_word
										if len(_a					) == 1:
													break
										else:
													lowerCamelCase      =   get_pairs(_a					)
							lowerCamelCase      =   """@@ """.join(_a					)
							lowerCamelCase      =   word[:-4]
							lowerCamelCase      =   word
							return word





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   []

							lowerCamelCase      =   re.findall(r"""\S+\n?"""						,					_a					)

							for token in words:
										split_tokens.extend(list(self.bpe(_a					).split(""" """					)					)					)
							return split_tokens





				def   _lowerCAmelCase							(			self						,					_a					):





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





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							return self.decoder.get(_a						,					self.unk_token					)





				def   _lowerCAmelCase							(			self						,					_a					):





							"""simple docstring"""
							lowerCamelCase      =   """ """.join(_a					).replace("""@@ """						,					""""""					).strip()
							return out_string





				def   _lowerCAmelCase							(			self						,					_a						,					_a = None					):





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

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

							lowerCamelCase      =   0
							with open(_a						,					"""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 _a					: 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      =   token_index
													writer.write(""" """.join(_a					) + """\n"""					)
													index += 1

							return vocab_file, merge_file

				# def decode(self, token_ids, skip_special_tokens=False, clean_up_tokenization_spaces=True):
				#     filtered_tokens = ' '.join(self.convert_ids_to_tokens(token_ids, skip_special_tokens=skip_special_tokens))
				#     tokens_generated_so_far = re.sub('(@@ )', '', string=filtered_tokens)
				#     tokens_generated_so_far = re.sub('(@@ ?$)', '', string=tokens_generated_so_far)
				#     return ''.join(tokens_generated_so_far)



