





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

import regex as re

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


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


_lowerCAmelCase				:   Union[str, Any]						       =						{'''vocab_file''': '''vocab.json''', '''merges_file''': '''merges.txt'''}

# See all BART models at https://huggingface.co/models?filter=bart
_lowerCAmelCase				:   Dict						       =						{
    '''vocab_file''': {
        '''facebook/bart-base''': '''https://huggingface.co/facebook/bart-base/resolve/main/vocab.json''',
        '''facebook/bart-large''': '''https://huggingface.co/facebook/bart-large/resolve/main/vocab.json''',
        '''facebook/bart-large-mnli''': '''https://huggingface.co/facebook/bart-large-mnli/resolve/main/vocab.json''',
        '''facebook/bart-large-cnn''': '''https://huggingface.co/facebook/bart-large-cnn/resolve/main/vocab.json''',
        '''facebook/bart-large-xsum''': '''https://huggingface.co/facebook/bart-large-xsum/resolve/main/vocab.json''',
        '''yjernite/bart_eli5''': '''https://huggingface.co/yjernite/bart_eli5/resolve/main/vocab.json''',
    },
    '''merges_file''': {
        '''facebook/bart-base''': '''https://huggingface.co/facebook/bart-base/resolve/main/merges.txt''',
        '''facebook/bart-large''': '''https://huggingface.co/facebook/bart-large/resolve/main/merges.txt''',
        '''facebook/bart-large-mnli''': '''https://huggingface.co/facebook/bart-large-mnli/resolve/main/merges.txt''',
        '''facebook/bart-large-cnn''': '''https://huggingface.co/facebook/bart-large-cnn/resolve/main/merges.txt''',
        '''facebook/bart-large-xsum''': '''https://huggingface.co/facebook/bart-large-xsum/resolve/main/merges.txt''',
        '''yjernite/bart_eli5''': '''https://huggingface.co/yjernite/bart_eli5/resolve/main/merges.txt''',
    },
}

_lowerCAmelCase				:   int						       =						{
    '''facebook/bart-base''': 1_024,
    '''facebook/bart-large''': 1_024,
    '''facebook/bart-large-mnli''': 1_024,
    '''facebook/bart-large-cnn''': 1_024,
    '''facebook/bart-large-xsum''': 1_024,
    '''yjernite/bart_eli5''': 1_024,
}

@lru_cache()
def   __snake_case    (       )			->    int:
   A_       :					List[str]        =      (
       list(range(ord("!"						)					,					ord("~"						) + 1						)						) + list(range(ord("¡"						)					,					ord("¬"						) + 1						)						) + list(range(ord("®"						)					,					ord("ÿ"						) + 1						)						)
   )
   A_       :					Optional[int]        =      bs[:]
   A_       :					Optional[Any]        =      0
   for b in range(2**8						):
      if b not in bs:
         bs.append(__lowerCamelCase						)
         cs.append(2**8 + n						)
         n += 1
   A_       :					Tuple        =      [chr(__lowerCamelCase						) for n in cs]
   return dict(zip(__lowerCamelCase					,					__lowerCamelCase						)						)

def   __snake_case    (       _lowerCAmelCase			:			Optional[int]						)			->    int:
   A_       :					Optional[Any]        =      set()
   A_       :					str        =      word[0]
   for char in word[1:]:
      pairs.add((prev_char, char)						)
      A_       :					Tuple        =      char
   return pairs





class        __magic_name__	(				A_				):





     """simple docstring"""



     __UpperCamelCase            =  VOCAB_FILES_NAMES
     __UpperCamelCase            =  PRETRAINED_VOCAB_FILES_MAP
     __UpperCamelCase            =  PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     __UpperCamelCase            =  ["input_ids", "attention_mask"]



     def __init__(				self							:Dict      ,   snake_case							:List[Any]      ,   snake_case							:Dict      ,   snake_case							:Union[str, Any]="replace"      ,   snake_case							:str="<s>"      ,   snake_case							:str="</s>"      ,   snake_case							:int="</s>"      ,   snake_case							:Any="<s>"      ,   snake_case							:str="<unk>"      ,   snake_case							:List[Any]="<pad>"      ,   snake_case							:str="<mask>"      ,   snake_case							:Optional[Any]=False      ,   **snake_case							:Optional[Any]      ,   ):





        '''simple docstring'''




        A_       :					str        =      AddedToken(snake_case__      ,   lstrip=snake_case__      ,   rstrip=snake_case__	) if isinstance(snake_case__      ,   snake_case__	) else bos_token
        A_       :					List[str]        =      AddedToken(snake_case__      ,   lstrip=snake_case__      ,   rstrip=snake_case__	) if isinstance(snake_case__      ,   snake_case__	) else eos_token
        A_       :					Optional[Any]        =      AddedToken(snake_case__      ,   lstrip=snake_case__      ,   rstrip=snake_case__	) if isinstance(snake_case__      ,   snake_case__	) else sep_token
        A_       :					Optional[int]        =      AddedToken(snake_case__      ,   lstrip=snake_case__      ,   rstrip=snake_case__	) if isinstance(snake_case__      ,   snake_case__	) else cls_token
        A_       :					Optional[Any]        =      AddedToken(snake_case__      ,   lstrip=snake_case__      ,   rstrip=snake_case__	) if isinstance(snake_case__      ,   snake_case__	) else unk_token
        A_       :					Any        =      AddedToken(snake_case__      ,   lstrip=snake_case__      ,   rstrip=snake_case__	) if isinstance(snake_case__      ,   snake_case__	) else pad_token

        # Mask token behave like a normal word, i.e. include the space before it
        A_       :					Tuple        =      AddedToken(snake_case__      ,   lstrip=snake_case__      ,   rstrip=snake_case__	) if isinstance(snake_case__      ,   snake_case__	) else mask_token

        super().__init__(
            errors=snake_case__      ,   bos_token=snake_case__      ,   eos_token=snake_case__      ,   unk_token=snake_case__      ,   sep_token=snake_case__      ,   cls_token=snake_case__      ,   pad_token=snake_case__      ,   mask_token=snake_case__      ,   add_prefix_space=snake_case__      ,   **snake_case__      ,   )

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

        # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
        A_       :					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     SCREAMING_SNAKE_CASE		(				self							:Optional[Any]	):





        '''simple docstring'''




        return len(self.encoder	)



     def     SCREAMING_SNAKE_CASE		(				self							:Optional[Any]	):





        '''simple docstring'''




        return dict(self.encoder      ,   **self.added_tokens_encoder	)



     def     SCREAMING_SNAKE_CASE		(				self							:List[str]      ,   snake_case							:List[Any]	):





        '''simple docstring'''




        if token in self.cache:
           return self.cache[token]
        A_       :					Optional[Any]        =      tuple(snake_case__	)
        A_       :					str        =      get_pairs(snake_case__	)

        if not pairs:
           return token

        while True:
           A_       :					Optional[int]        =      min(snake_case__      ,   key=lambda snake_case	: self.bpe_ranks.get(snake_case__      ,   float("inf"	)	)	)
           if bigram not in self.bpe_ranks:
              break
           A_       :					str        =      bigram
           A_       :					List[str]        =      []
           A_       :					Optional[int]        =      0
           while i < len(snake_case__	):
              try:
                 A_       :					List[Any]        =      word.index(snake_case__      ,   snake_case__	)
              except ValueError:
                 new_word.extend(word[i:]	)
                 break
              else:
                 new_word.extend(word[i:j]	)
                 A_       :					Dict        =      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
           A_       :					Any        =      tuple(snake_case__	)
           A_       :					int        =      new_word
           if len(snake_case__	) == 1:
              break
           else:
              A_       :					List[str]        =      get_pairs(snake_case__	)
        A_       :					Tuple        =      " ".join(snake_case__	)
        A_       :					List[Any]        =      word
        return word



     def     SCREAMING_SNAKE_CASE		(				self							:Optional[Any]      ,   snake_case							:Dict	):





        '''simple docstring'''




        A_       :					List[str]        =      []
        for token in re.findall(self.pat      ,   snake_case__	):
           A_       :					List[str]        =      "".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     SCREAMING_SNAKE_CASE		(				self							:Optional[Any]      ,   snake_case							:Any	):





        '''simple docstring'''




        return self.encoder.get(snake_case__      ,   self.encoder.get(self.unk_token	)	)



     def     SCREAMING_SNAKE_CASE		(				self							:Optional[int]      ,   snake_case							:Optional[int]	):





        '''simple docstring'''




        return self.decoder.get(snake_case__	)



     def     SCREAMING_SNAKE_CASE		(				self							:Tuple      ,   snake_case							:Tuple	):





        '''simple docstring'''




        A_       :					str        =      "".join(snake_case__	)
        A_       :					Union[str, Any]        =      bytearray([self.byte_decoder[c] for c in text]	).decode("utf-8"      ,   errors=self.errors	)
        return text



     def     SCREAMING_SNAKE_CASE		(				self							:Tuple      ,   snake_case							:str      ,   snake_case							:Optional[str] = None	):





        '''simple docstring'''




        if not os.path.isdir(snake_case__	):
           logger.error(f"Vocabulary path ({save_directory}) should be a directory"	)
           return
        A_       :					List[Any]        =      os.path.join(
            snake_case__      ,   (filename_prefix + "-" if filename_prefix else "") + VOCAB_FILES_NAMES["vocab_file"]	)
        A_       :					str        =      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"	)

        A_       :					List[str]        =      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 snake_case	: 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!"	)
                 A_       :					Optional[Any]        =      token_index
              writer.write(" ".join(snake_case__	) + "\n"	)
              index += 1

        return vocab_file, merge_file



     def     SCREAMING_SNAKE_CASE		(				self							:List[str]      ,   snake_case							:List[int]      ,   snake_case							:Optional[List[int]] = None	):





        '''simple docstring'''




        if token_ids_a is None:
           return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
        A_       :					int        =      [self.cls_token_id]
        A_       :					Optional[Any]        =      [self.sep_token_id]
        return cls + token_ids_a + sep + sep + token_ids_a + sep



     def     SCREAMING_SNAKE_CASE		(				self							:str      ,   snake_case							:List[int]      ,   snake_case							:Optional[List[int]] = None      ,   snake_case							:bool = False	):





        '''simple docstring'''




        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 token_ids_a is None:
           return [1] + ([0] * len(snake_case__	)) + [1]
        return [1] + ([0] * len(snake_case__	)) + [1, 1] + ([0] * len(snake_case__	)) + [1]



     def     SCREAMING_SNAKE_CASE		(				self							:Optional[Any]      ,   snake_case							:List[int]      ,   snake_case							:Optional[List[int]] = None	):





        '''simple docstring'''




        A_       :					Optional[int]        =      [self.sep_token_id]
        A_       :					Optional[Any]        =      [self.cls_token_id]

        if token_ids_a is None:
           return len(cls + token_ids_a + sep	) * [0]
        return len(cls + token_ids_a + sep + sep + token_ids_a + sep	) * [0]





     def     SCREAMING_SNAKE_CASE		(				self							:Dict      ,   snake_case							:Any      ,   snake_case							:Dict=False      ,   **snake_case							:Dict	):





        '''simple docstring'''




        A_       :					Optional[Any]        =      kwargs.pop("add_prefix_space"      ,   self.add_prefix_space	)
        if (is_split_into_words or add_prefix_space) and (len(snake_case__	) > 0 and not text[0].isspace()):
           A_       :					Union[str, Any]        =      " " + text
        return (text, kwargs)


from ..utils import (
    OptionalDependencyNotAvailable,
    is_flax_available,
    is_scipy_available,
    is_torch_available,
    is_torchsde_available,
)


try:
   if not is_torch_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   from ..utils.dummy_pt_objects import *  # noqa F403
else:
   from .scheduling_consistency_models import CMStochasticIterativeScheduler
   from .scheduling_ddim import DDIMScheduler
   from .scheduling_ddim_inverse import DDIMInverseScheduler
   from .scheduling_ddim_parallel import DDIMParallelScheduler
   from .scheduling_ddpm import DDPMScheduler
   from .scheduling_ddpm_parallel import DDPMParallelScheduler
   from .scheduling_deis_multistep import DEISMultistepScheduler
   from .scheduling_dpmsolver_multistep import DPMSolverMultistepScheduler
   from .scheduling_dpmsolver_multistep_inverse import DPMSolverMultistepInverseScheduler
   from .scheduling_dpmsolver_singlestep import DPMSolverSinglestepScheduler
   from .scheduling_euler_ancestral_discrete import EulerAncestralDiscreteScheduler
   from .scheduling_euler_discrete import EulerDiscreteScheduler
   from .scheduling_heun_discrete import HeunDiscreteScheduler
   from .scheduling_ipndm import IPNDMScheduler
   from .scheduling_k_dpm_2_ancestral_discrete import KDPMaAncestralDiscreteScheduler
   from .scheduling_k_dpm_2_discrete import KDPMaDiscreteScheduler
   from .scheduling_karras_ve import KarrasVeScheduler
   from .scheduling_pndm import PNDMScheduler
   from .scheduling_repaint import RePaintScheduler
   from .scheduling_sde_ve import ScoreSdeVeScheduler
   from .scheduling_sde_vp import ScoreSdeVpScheduler
   from .scheduling_unclip import UnCLIPScheduler
   from .scheduling_unipc_multistep import UniPCMultistepScheduler
   from .scheduling_utils import KarrasDiffusionSchedulers, SchedulerMixin
   from .scheduling_vq_diffusion import VQDiffusionScheduler

try:
   if not is_flax_available():
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   from ..utils.dummy_flax_objects import *  # noqa F403
else:
   from .scheduling_ddim_flax import FlaxDDIMScheduler
   from .scheduling_ddpm_flax import FlaxDDPMScheduler
   from .scheduling_dpmsolver_multistep_flax import FlaxDPMSolverMultistepScheduler
   from .scheduling_karras_ve_flax import FlaxKarrasVeScheduler
   from .scheduling_lms_discrete_flax import FlaxLMSDiscreteScheduler
   from .scheduling_pndm_flax import FlaxPNDMScheduler
   from .scheduling_sde_ve_flax import FlaxScoreSdeVeScheduler
   from .scheduling_utils_flax import (
       FlaxKarrasDiffusionSchedulers,
       FlaxSchedulerMixin,
       FlaxSchedulerOutput,
       broadcast_to_shape_from_left,
   )


try:
   if not (is_torch_available() and is_scipy_available()):
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   from ..utils.dummy_torch_and_scipy_objects import *  # noqa F403
else:
   from .scheduling_lms_discrete import LMSDiscreteScheduler

try:
   if not (is_torch_available() and is_torchsde_available()):
      raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
   from ..utils.dummy_torch_and_torchsde_objects import *  # noqa F403
else:
   from .scheduling_dpmsolver_sde import DPMSolverSDEScheduler





