
"""simple docstring"""


from abc import ABC, abstractmethod
from typing import List, Optional







class  lowerCAmelCase_ (					_lowercase			):



						'''simple docstring'''


						def __init__(       self   :					Union[str, Any]					)   ->      Tuple:
							# test for the above condition
							self.test()


						def   _SCREAMING_SNAKE_CASE					(       self   :					Union[str, Any]					)   ->      Any:
							A        =      0
							A        =      False
							while not completed:
								if counter == 1:
									self.reset()
								A        =      self.advance()
								if not self.does_advance(A_					):
									raise Exception(
									    'Custom Constraint is not defined correctly. self.does_advance(self.advance()) must be true.'					)

								A      ,					A      ,					A        =      self.update(A_					)
								counter += 1

								if counter > 1_0000:
									raise Exception('update() does not fulfill the constraint.'					)

							if self.remaining() != 0:
								raise Exception('Custom Constraint is not defined correctly.'					)


						@abstractmethod
						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[int]					)   ->      List[str]:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'					)


						@abstractmethod
						def   _SCREAMING_SNAKE_CASE					(       self   :					Union[str, Any]     ,A_   :					int					)   ->      Optional[int]:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'					)


						@abstractmethod
						def   _SCREAMING_SNAKE_CASE					(       self   :					Dict     ,A_   :					int					)   ->      List[str]:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'					)


						@abstractmethod
						def   _SCREAMING_SNAKE_CASE					(       self   :					int					)   ->      List[str]:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'					)


						@abstractmethod
						def   _SCREAMING_SNAKE_CASE					(       self   :					int					)   ->      Optional[Any]:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'					)




						@abstractmethod
						def   _SCREAMING_SNAKE_CASE					(       self   :					List[str]     ,A_   :					Tuple=False					)   ->      Tuple:
							raise NotImplementedError(
							    F'{self.__class__} is an abstract class. Only classes inheriting this class can be called.'					)







class  lowerCAmelCase_ (					_lowercase			):



						'''simple docstring'''


						def __init__(       self   :					int     ,A_   :					List[int]					)   ->      Dict:
							super(A_     ,self					).__init__()

							if not isinstance(A_     ,A_					) or len(A_					) == 0:
								raise ValueError(F'`token_ids` has to be a non-empty list, but is {token_ids}.'					)
							if any((not isinstance(A_     ,A_					) or token_id < 0) for token_id in token_ids					):
								raise ValueError(F'Each list in `token_ids` has to be a list of positive integers, but is {token_ids}.'					)

							A        =      token_ids

							A        =      len(self.token_ids					)
							A        =      -1  # the index of the currently fulfilled step
							A        =      False


						def   _SCREAMING_SNAKE_CASE					(       self   :					int					)   ->      Union[str, Any]:
							if self.completed:
								return None
							return self.token_ids[self.fulfilled_idx + 1]


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[str]     ,A_   :					int					)   ->      Union[str, Any]:
							if not isinstance(A_     ,A_					):
								raise ValueError(F'`token_id` has to be an `int`, but is {token_id} of type {type(A_					)}'					)

							if self.completed:
								return False

							return token_id == self.token_ids[self.fulfilled_idx + 1]


						def   _SCREAMING_SNAKE_CASE					(       self   :					str     ,A_   :					int					)   ->      List[str]:
							if not isinstance(A_     ,A_					):
								raise ValueError(F'`token_id` has to be an `int`, but is {token_id} of type {type(A_					)}'					)

							A        =      False
							A        =      False
							A        =      False

							if self.does_advance(A_					):
								self.fulfilled_idx += 1
								A        =      True
								if self.fulfilled_idx == (self.seqlen - 1):
									A        =      True
								A        =      completed
							else:
								# failed to make progress.
								A        =      True
								self.reset()
							return stepped, completed, reset


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[Any]					)   ->      Dict:
							A        =      False
							A        =      0


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[Any]					)   ->      List[Any]:
							return self.seqlen - (self.fulfilled_idx + 1)




						def   _SCREAMING_SNAKE_CASE					(       self   :					Any     ,A_   :					Union[str, Any]=False					)   ->      Any:
							A        =      PhrasalConstraint(self.token_ids					)

							if stateful:
								A        =      self.seqlen
								A        =      self.fulfilled_idx
								A        =      self.completed

							return new_constraint







class  lowerCAmelCase_ :



						'''simple docstring'''


						def __init__(       self   :					str     ,A_   :					List[List[int]]     ,A_   :					Union[str, Any]=True					)   ->      Union[str, Any]:
							A        =      max([len(A_					) for one in nested_token_ids]					)

							A        =      {}
							for token_ids in nested_token_ids:
								A        =      root
								for tidx, token_id in enumerate(A_					):
									if token_id not in level:
										A        =      {}

									A        =      level[token_id]

							if no_subsets and self.has_subsets(A_     ,A_					):
								raise ValueError(
								    'Each list in `nested_token_ids` can\'t be a complete subset of another list, but is'
								    F' {nested_token_ids}.'					)

							A        =      root


						def   _SCREAMING_SNAKE_CASE					(       self   :					Any     ,A_   :					Tuple					)   ->      List[str]:
							A        =      self.trie

							for current_token in current_seq:
								A        =      start[current_token]

							A        =      list(start.keys()					)

							return next_tokens


						def   _SCREAMING_SNAKE_CASE					(       self   :					Tuple     ,A_   :					Optional[int]					)   ->      List[Any]:
							A        =      self.next_tokens(A_					)

							return len(A_					) == 0


						def   _SCREAMING_SNAKE_CASE					(       self   :					Tuple     ,A_   :					Union[str, Any]					)   ->      Any:
							A        =      list(root.values()					)
							if len(A_					) == 0:
								return 1
							else:
								return sum([self.count_leaves(A_					) for nn in next_nodes]					)




						def   _SCREAMING_SNAKE_CASE					(       self   :					Union[str, Any]     ,A_   :					Tuple     ,A_   :					Tuple					)   ->      Optional[Any]:
							A        =      self.count_leaves(A_					)
							return len(A_					) != leaf_count







class  lowerCAmelCase_ (					_lowercase			):



						'''simple docstring'''


						def __init__(       self   :					Any     ,A_   :					List[List[int]]					)   ->      Optional[Any]:
							super(A_     ,self					).__init__()

							if not isinstance(A_     ,A_					) or len(A_					) == 0:
								raise ValueError(F'`nested_token_ids` has to be a non-empty list, but is {nested_token_ids}.'					)
							if any(not isinstance(A_     ,A_					) for token_ids in nested_token_ids					):
								raise ValueError(F'`nested_token_ids` has to be a list of lists, but is {nested_token_ids}.'					)
							if any(
							    any((not isinstance(A_     ,A_					) or token_id < 0) for token_id in token_ids					)
							    for token_ids in nested_token_ids					):
								raise ValueError(
								    F'Each list in `nested_token_ids` has to be a list of positive integers, but is {nested_token_ids}.'					)

							A        =      DisjunctiveTrie(A_					)
							A        =      nested_token_ids

							A        =      self.trie.max_height
							A        =      []
							A        =      False


						def   _SCREAMING_SNAKE_CASE					(       self   :					Dict					)   ->      Dict:
							A        =      self.trie.next_tokens(self.current_seq					)

							if len(A_					) == 0:
								return None
							else:
								return token_list


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[Any]     ,A_   :					int					)   ->      Any:
							if not isinstance(A_     ,A_					):
								raise ValueError(F'`token_id` is supposed to be type `int`, but is {token_id} of type {type(A_					)}'					)

							A        =      self.trie.next_tokens(self.current_seq					)

							return token_id in next_tokens


						def   _SCREAMING_SNAKE_CASE					(       self   :					Tuple     ,A_   :					int					)   ->      Dict:
							if not isinstance(A_     ,A_					):
								raise ValueError(F'`token_id` is supposed to be type `int`, but is {token_id} of type {type(A_					)}'					)

							A        =      False
							A        =      False
							A        =      False

							if self.does_advance(A_					):
								self.current_seq.append(A_					)
								A        =      True
							else:
								A        =      True
								self.reset()

							A        =      self.trie.reached_leaf(self.current_seq					)
							A        =      completed

							return stepped, completed, reset


						def   _SCREAMING_SNAKE_CASE					(       self   :					int					)   ->      str:
							A        =      False
							A        =      []


						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[int]					)   ->      Dict:
							if self.completed:
								# since this can be completed without reaching max height
								return 0
							else:
								return self.seqlen - len(self.current_seq					)




						def   _SCREAMING_SNAKE_CASE					(       self   :					List[str]     ,A_   :					Optional[Any]=False					)   ->      Any:
							A        =      DisjunctiveConstraint(self.token_ids					)

							if stateful:
								A        =      self.seqlen
								A        =      self.current_seq
								A        =      self.completed

							return new_constraint







class  lowerCAmelCase_ :



						'''simple docstring'''


						def __init__(       self   :					Optional[Any]     ,A_   :					List[Constraint]					)   ->      Optional[int]:
							A        =      constraints

							# max # of steps required to fulfill a given constraint
							A        =      max([c.seqlen for c in constraints]					)
							A        =      len(A_					)
							A        =      False

							self.init_state()


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[Any]					)   ->      List[Any]:
							A        =      []
							A        =      None
							A        =      [constraint.copy(stateful=A_					) for constraint in self.constraints]


						def   _SCREAMING_SNAKE_CASE					(       self   :					Tuple					)   ->      Any:
							A        =      0
							if self.inprogress_constraint:
								# extra points for having a constraint mid-fulfilled
								add += self.max_seqlen - self.inprogress_constraint.remaining()

							return (len(self.complete_constraints					) * self.max_seqlen) + add


						def   _SCREAMING_SNAKE_CASE					(       self   :					List[Any]					)   ->      Optional[int]:
							A        =      []
							if self.inprogress_constraint is None:
								for constraint in self.pending_constraints:  # "pending" == "unfulfilled yet"
									A        =      constraint.advance()
									if isinstance(A_     ,A_					):
										token_list.append(A_					)
									elif isinstance(A_     ,A_					):
										token_list.extend(A_					)
							else:
								A        =      self.inprogress_constraint.advance()
								if isinstance(A_     ,A_					):
									token_list.append(A_					)
								elif isinstance(A_     ,A_					):
									token_list.extend(A_					)

							if len(A_					) == 0:
								return None
							else:
								return token_list


						def   _SCREAMING_SNAKE_CASE					(       self   :					Dict     ,A_   :					Optional[List[int]]					)   ->      List[Any]:
							self.init_state()

							if token_ids is not None:
								for token in token_ids:
									# completes or steps **one** constraint
									A      ,					A        =      self.add(A_					)

									# the entire list of constraints are fulfilled
									if self.completed:
										break


						def   _SCREAMING_SNAKE_CASE					(       self   :					Union[str, Any]     ,A_   :					int					)   ->      Tuple:
							if not isinstance(A_     ,A_					):
								raise ValueError(F'`token_id` should be an `int`, but is `{token_id}`.'					)

							A      ,					A        =      False, False

							if self.completed:
								A        =      True
								A        =      False
								return complete, stepped

							if self.inprogress_constraint is not None:
								# In the middle of fulfilling a constraint. If the `token_id` *does* makes an incremental progress to current
								# job, simply update the state

								A      ,					A      ,					A        =      self.inprogress_constraint.update(A_					)
								if reset:
									# 1. If the next token breaks the progress, then we must restart.
									#     e.g. constraint = "I love pies" and sequence so far is "I love" but `token_id` == "books".

									#     But that doesn't mean we self.init_state(), since we only reset the state for this particular
									#     constraint, not the full list of constraints.

									self.pending_constraints.append(self.inprogress_constraint.copy(stateful=A_					)					)
									A        =      None

								if complete:
									# 2. If the next token completes the constraint, move it to completed list, set
									#     inprogress to None. If there are no pending constraints either, then this full list of constraints
									#     is complete.

									self.complete_constraints.append(self.inprogress_constraint					)
									A        =      None

									if len(self.pending_constraints					) == 0:
										# we're done!
										A        =      True

							else:
								# Not in the middle of fulfilling a constraint. So does this `token_id` helps us step towards any of our list
								# of constraints?

								for cidx, pending_constraint in enumerate(self.pending_constraints					):
									if pending_constraint.does_advance(A_					):
										A      ,					A      ,					A        =      pending_constraint.update(A_					)

										if not stepped:
											raise Exception(
											    '`constraint.update(token_id)` is not yielding incremental progress, '
											    'even though `constraint.does_advance(token_id)` is true.'					)

										if complete:
											self.complete_constraints.append(A_					)
											A        =      None

										if not complete and stepped:
											A        =      pending_constraint

										if complete or stepped:
											# If we made any progress at all, then it's at least not a "pending constraint".

											A        =      (
											    self.pending_constraints[:cidx] + self.pending_constraints[cidx + 1 :]
											)

											if len(self.pending_constraints					) == 0 and self.inprogress_constraint is None:
												# If there's no longer any pending after this and no inprogress either, then we must be
												# complete.

												A        =      True

											break  # prevent accidentally stepping through multiple constraints with just one token.

							return complete, stepped




						def   _SCREAMING_SNAKE_CASE					(       self   :					Optional[Any]     ,A_   :					Dict=True					)   ->      Optional[int]:
							A        =      ConstraintListState(self.constraints					)  # we actually never though self.constraints objects
							# throughout this process. So it's at initialization state.

							if stateful:
								A        =      [
								    constraint.copy(stateful=A_					) for constraint in self.complete_constraints
								]
								if self.inprogress_constraint is not None:
									A        =      self.inprogress_constraint.copy(stateful=A_					)
								A        =      [constraint.copy() for constraint in self.pending_constraints]

							return new_state




import os
from shutil import copyfile
from typing import Any, Dict, List, Optional, Tuple

import sentencepiece as spm

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


SCREAMING_SNAKE_CASE_:List[Any]             =						"""▁"""

SCREAMING_SNAKE_CASE_:int             =						{"""vocab_file""": """spiece.model"""}

SCREAMING_SNAKE_CASE_:Any             =						{
    """vocab_file""": {"""google/pegasus-xsum""": """https://huggingface.co/google/pegasus-xsum/resolve/main/spiece.model"""}
}

SCREAMING_SNAKE_CASE_:Optional[Any]             =						{
    """google/pegasus-xsum""": 512,
}


SCREAMING_SNAKE_CASE_:Tuple             =						logging.get_logger(__name__)






class SCREAMING_SNAKE_CASE__     (		SCREAMING_SNAKE_CASE__    ):




    '''simple docstring'''






    __lowerCamelCase			: Union[str, Any]					=      VOCAB_FILES_NAMES

    __lowerCamelCase			: List[Any]					=      VOCAB_FILES_NAMES
    __lowerCamelCase			: Optional[Any]					=      PRETRAINED_VOCAB_FILES_MAP
    __lowerCamelCase			: str					=      PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
    __lowerCamelCase			: str					=      ["input_ids", "attention_mask"]






    def __init__(       self,		lowerCamelCase__,		lowerCamelCase__="<pad>",		lowerCamelCase__="</s>",		lowerCamelCase__="<unk>",		lowerCamelCase__="<mask_2>",		lowerCamelCase__="<mask_1>",		lowerCamelCase__=None,		lowerCamelCase__=103,		lowerCamelCase__ = None,		**lowerCamelCase__,		):
        A :       int          =   offset
        if additional_special_tokens is not None:
            if not isinstance(lowerCamelCase__,		lowerCamelCase__		):
                raise TypeError(
                    f'''additional_special_tokens should be of type {type(lowerCamelCase__		)}, but is'''
                    f''' {type(lowerCamelCase__		)}'''		)

            A :       Dict          =   (
                ([mask_token_sent] + additional_special_tokens)
                if mask_token_sent not in additional_special_tokens and mask_token_sent is not None
                else additional_special_tokens
            )
            # fill additional tokens with ..., <unk_token_102> in case not all additional tokens are already taken
            additional_special_tokens_extended += [
                f'''<unk_{i}>''' for i in range(len(lowerCamelCase__		),		self.offset - 1		)
            ]

            if len(set(lowerCamelCase__		)		) != len(lowerCamelCase__		):
                raise ValueError(
                    """Please make sure that the provided additional_special_tokens do not contain an incorrectly"""
                    f''' shifted list of <unk_x> tokens. Found {additional_special_tokens_extended}.'''		)
            A :       int          =   additional_special_tokens_extended
        else:
            A :       Optional[int]          =   [mask_token_sent] if mask_token_sent is not None else []
            additional_special_tokens += [f'''<unk_{i}>''' for i in range(2,		self.offset		)]

        A :       Any          =   {} if sp_model_kwargs is None else sp_model_kwargs

        super().__init__(
            eos_token=lowerCamelCase__,		unk_token=lowerCamelCase__,		mask_token=lowerCamelCase__,		pad_token=lowerCamelCase__,		mask_token_sent=lowerCamelCase__,		offset=lowerCamelCase__,		additional_special_tokens=lowerCamelCase__,		sp_model_kwargs=self.sp_model_kwargs,		**lowerCamelCase__,		)
        A :       Union[str, Any]          =   mask_token_sent
        A :       Optional[Any]          =   vocab_file
        A :       Optional[int]          =   spm.SentencePieceProcessor(**self.sp_model_kwargs		)
        self.sp_model.Load(lowerCamelCase__		)

        # add special tokens to encoder dict
        A :       Dict[int, str]          =   {
            0: self.pad_token,
            1: self.eos_token,
        }

        if self.mask_token_sent is not None:
            self.encoder.update(
                {
                    2: self.mask_token_sent,
                    3: self.mask_token,
                }		)

        if self.offset > 0:
            # entries 2-104 are only used for pretraining and called <mask_1>, <mask_2>, unk_2, ...unk_102
            # mask_token_sent is already added to list -> so start at 1
            self.encoder.update({i + 3: additional_special_tokens[i] for i in range(1,		self.offset - 1		)}		)

        A :       Dict[str, int]          =   {v: k for k, v in self.encoder.items()}






    @property
    def 			_lowerCAmelCase					(       self		):
        return len(self.sp_model		) + self.offset






    def 			_lowerCAmelCase					(       self		):
        A :       Optional[int]          =   {self.convert_ids_to_tokens(lowerCamelCase__		): i for i in range(self.vocab_size		)}
        vocab.update(self.added_tokens_encoder		)
        return vocab






    def __getstate__(       self		):
        A :       List[Any]          =   self.__dict__.copy()
        A :       Union[str, Any]          =   None
        return state






    def __setstate__(       self,		lowerCamelCase__		):
        A :       List[Any]          =   d

        # for backward compatibility
        if not hasattr(self,		"""sp_model_kwargs"""		):
            A :       int          =   {}

        A :       Tuple          =   spm.SentencePieceProcessor(**self.sp_model_kwargs		)
        self.sp_model.Load(self.vocab_file		)






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__		):
        return self.sp_model.encode(lowerCamelCase__,		out_type=lowerCamelCase__		)






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__		):
        if token in self.decoder:
            return self.decoder[token]
        elif token in self.added_tokens_decoder:
            return self.added_tokens_decoder[token]
        A :       List[str]          =   self.sp_model.piece_to_id(lowerCamelCase__		)
        return sp_id + self.offset






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__		):
        if index in self.encoder:
            return self.encoder[index]
        elif index in self.added_tokens_encoder:
            return self.added_tokens_encoder[index]
        else:
            A :       Dict          =   self.sp_model.IdToPiece(index - self.offset		)
        return token






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__		):
        A :       Optional[int]          =   []
        A :       Optional[Any]          =   """"""
        for token in tokens:
            # make sure that special tokens are not decoded using sentencepiece model
            if token in self.all_special_tokens:
                out_string += self.sp_model.decode(lowerCamelCase__		) + token
                A :       int          =   []
            else:
                current_sub_tokens.append(lowerCamelCase__		)
        out_string += self.sp_model.decode(lowerCamelCase__		)
        return out_string.strip()






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__=False		):
        return 1






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__		):
        A :       Optional[int]          =   set(self.all_special_ids		)  # call it once instead of inside list comp
        all_special_ids.remove(self.unk_token_id		)  # <unk> is only sometimes special

        return [1 if x in all_special_ids else 0 for x in seq]






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__ = None,		lowerCamelCase__ = False		):
        if already_has_special_tokens:
            return self._special_token_mask(lowerCamelCase__		)
        elif token_ids_a is None:
            return self._special_token_mask(lowerCamelCase__		) + [1]
        else:
            return self._special_token_mask(token_ids_a + token_ids_a		) + [1]






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__=None		):
        if token_ids_a is None:
            return token_ids_a + [self.eos_token_id]
        # We don't expect to process pairs, but leave the pair logic for API consistency
        return token_ids_a + token_ids_a + [self.eos_token_id]






    def 			_lowerCAmelCase					(       self,		lowerCamelCase__,		lowerCamelCase__ = None		):
        if not os.path.isdir(lowerCamelCase__		):
            logger.error(f'''Vocabulary path ({save_directory}) should be a directory'''		)
            return
        A :       Any          =   os.path.join(
            lowerCamelCase__,		(filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]		)

        if os.path.abspath(self.vocab_file		) != os.path.abspath(lowerCamelCase__		) and os.path.isfile(self.vocab_file		):
            copyfile(self.vocab_file,		lowerCamelCase__		)
        elif not os.path.isfile(self.vocab_file		):
            with open(lowerCamelCase__,		"""wb"""		) as fi:
                A :       Dict          =   self.sp_model.serialized_model_proto()
                fi.write(lowerCamelCase__		)

        return (out_vocab_file,)


