




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


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


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

# See all MVP models at https://huggingface.co/models?filter=mvp
SCREAMING_SNAKE_CASE_    :       str       =							{
    '''vocab_file''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/vocab.json''',
    },
    '''added_tokens.json''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/added_tokens.json''',
    },
    '''merges_file''': {
        '''RUCAIBox/mvp''': '''https://huggingface.co/RUCAIBox/mvp/resolve/main/merges.txt''',
    },
}

SCREAMING_SNAKE_CASE_    :       str       =							{
    '''RUCAIBox/mvp''': 10_24,
}
@lru_cache()
def 							__lowerCAmelCase			():
    a__  			=							(
        list(range(ord("""!"""		) ,     ord("""~"""		) + 1		)		) + list(range(ord("""¡"""		) ,     ord("""¬"""		) + 1		)		) + list(range(ord("""®"""		) ,     ord("""ÿ"""		) + 1		)		)
    )
    a__  			=							bs[:]
    a__  			=							0
    for b in range(2**8		):
        if b not in bs:
            bs.append(UpperCAmelCase_		)
            cs.append(2**8 + n		)
            n += 1
    a__  			=							[chr(UpperCAmelCase_		) for n in cs]
    return dict(zip(UpperCAmelCase_ ,     UpperCAmelCase_		)		)
def 							__lowerCAmelCase			(UpperCAmelCase_		):
    a__  			=							set()
    a__  			=							word[0]
    for char in word[1:]:
        pairs.add((prev_char, char)		)
        a__  			=							char
    return pairs







class 		__UpperCAmelCase(		lowercase__			):





     """simple docstring"""



     __UpperCAmelCase						:			List[Any]							=     VOCAB_FILES_NAMES
     __UpperCAmelCase						:			List[str]							=     PRETRAINED_VOCAB_FILES_MAP
     __UpperCAmelCase						:			Tuple							=     PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
     __UpperCAmelCase						:			Any							=     ["""input_ids""", """attention_mask"""]

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






         """simple docstring"""
         a__  			=							AddedToken(A    ,					lstrip=A    ,					rstrip=A     ) if isinstance(A    ,					A     ) else bos_token
         a__  			=							AddedToken(A    ,					lstrip=A    ,					rstrip=A     ) if isinstance(A    ,					A     ) else eos_token
         a__  			=							AddedToken(A    ,					lstrip=A    ,					rstrip=A     ) if isinstance(A    ,					A     ) else sep_token
         a__  			=							AddedToken(A    ,					lstrip=A    ,					rstrip=A     ) if isinstance(A    ,					A     ) else cls_token
         a__  			=							AddedToken(A    ,					lstrip=A    ,					rstrip=A     ) if isinstance(A    ,					A     ) else unk_token
         a__  			=							AddedToken(A    ,					lstrip=A    ,					rstrip=A     ) if isinstance(A    ,					A     ) else pad_token

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

         super().__init__(
             errors=A    ,					bos_token=A    ,					eos_token=A    ,					unk_token=A    ,					sep_token=A    ,					cls_token=A    ,					pad_token=A    ,					mask_token=A    ,					add_prefix_space=A    ,					**A    ,					)
         with open(A    ,					encoding="""utf-8"""     ) as vocab_handle:
             a__  			=							json.load(A     )
         a__  			=							{v: k for k, v in self.encoder.items()}
         a__  			=							errors  # how to handle errors in decoding
         a__  			=							bytes_to_unicode()
         a__  			=							{v: k for k, v in self.byte_encoder.items()}
         with open(A    ,					encoding="""utf-8"""     ) as merges_handle:
             a__  			=							merges_handle.read().split("""\n"""     )[1:-1]
         a__  			=							[tuple(merge.split()     ) for merge in bpe_merges]
         a__  			=							dict(zip(A    ,					range(len(A     )     )     )     )
         a__  			=							{}
         a__  			=							add_prefix_space

         # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
         a__  			=							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			:							List[Any]     )      ->      Optional[int]:






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

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Any     )      ->      List[Any]:






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

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Dict    ,					A			:							str     )      ->      str:






         """simple docstring"""
         if token in self.cache:
             return self.cache[token]
         a__  			=							tuple(A     )
         a__  			=							get_pairs(A     )

         if not pairs:
             return token

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

     def 					__SCREAMING_SNAKE_CASE		(   self			:							List[Any]    ,					A			:							int     )      ->      List[str]:






         """simple docstring"""
         a__  			=							[]
         for token in re.findall(self.pat    ,					A     ):
             a__  			=							"""""".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(A     ).split(""" """     )     )
         return bpe_tokens

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Union[str, Any]    ,					A			:							List[str]     )      ->      Union[str, Any]:






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

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Tuple    ,					A			:							int     )      ->      Union[str, Any]:






         """simple docstring"""
         return self.decoder.get(A     )

     def 					__SCREAMING_SNAKE_CASE		(   self			:							int    ,					A			:							Union[str, Any]     )      ->      Any:






         """simple docstring"""
         a__  			=							"""""".join(A     )
         a__  			=							bytearray([self.byte_decoder[c] for c in text]     ).decode("""utf-8"""    ,					errors=self.errors     )
         return text

     def 					__SCREAMING_SNAKE_CASE		(   self			:							List[str]    ,					A			:							str    ,					A			:							Optional[str] = None     )      ->      Tuple[str]:






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

         a__  			=							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!"""     )
                     a__  			=							token_index
                 writer.write(""" """.join(A     ) + """\n"""     )
                 index += 1

         return vocab_file, merge_file

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Optional[Any]    ,					A			:							List[int]    ,					A			:							Optional[List[int]] = None     )      ->      List[int]:






         """simple docstring"""
         if token_ids_a is None:
             return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
         a__  			=							[self.cls_token_id]
         a__  			=							[self.sep_token_id]
         return cls + token_ids_a + sep + sep + token_ids_a + sep

     def 					__SCREAMING_SNAKE_CASE		(   self			:							Union[str, Any]    ,					A			:							List[int]    ,					A			:							Optional[List[int]] = None    ,					A			:							bool = False     )      ->      List[int]:






         """simple docstring"""
         if already_has_special_tokens:
             return super().get_special_tokens_mask(
                 token_ids_a=A    ,					token_ids_a=A    ,					already_has_special_tokens=A     )

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

     def 					__SCREAMING_SNAKE_CASE		(   self			:							int    ,					A			:							List[int]    ,					A			:							Optional[List[int]] = None     )      ->      List[int]:






         """simple docstring"""
         a__  			=							[self.sep_token_id]
         a__  			=							[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			:							Optional[int]    ,					A			:							List[str]    ,					A			:							Dict=False    ,					**A			:							Tuple     )      ->      Optional[int]:






         """simple docstring"""
         a__  			=							kwargs.pop("""add_prefix_space"""    ,					self.add_prefix_space     )
         if (is_split_into_words or add_prefix_space) and (len(A     ) > 0 and not text[0].isspace()):
             a__  			=							""" """ + text
         return (text, kwargs)












# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from typing import TYPE_CHECKING

from ..utils import (
    OptionalDependencyNotAvailable,
    _LazyModule,
    is_torch_available,
)


SCREAMING_SNAKE_CASE_    :       int       =							{
    '''agents''': ['''Agent''', '''AzureOpenAiAgent''', '''HfAgent''', '''LocalAgent''', '''OpenAiAgent'''],
    '''base''': ['''PipelineTool''', '''RemoteTool''', '''Tool''', '''launch_gradio_demo''', '''load_tool'''],
}

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 SCREAMING_SNAKE_CASE_    :       int       =							['''DocumentQuestionAnsweringTool''']
 SCREAMING_SNAKE_CASE_    :       List[Any]       =							['''ImageCaptioningTool''']
 SCREAMING_SNAKE_CASE_    :       Union[str, Any]       =							['''ImageQuestionAnsweringTool''']
 SCREAMING_SNAKE_CASE_    :       str       =							['''ImageSegmentationTool''']
 SCREAMING_SNAKE_CASE_    :       Any       =							['''SpeechToTextTool''']
 SCREAMING_SNAKE_CASE_    :       Union[str, Any]       =							['''TextClassificationTool''']
 SCREAMING_SNAKE_CASE_    :       List[Any]       =							['''TextQuestionAnsweringTool''']
 SCREAMING_SNAKE_CASE_    :       Dict       =							['''TextSummarizationTool''']
 SCREAMING_SNAKE_CASE_    :       Optional[int]       =							['''TextToSpeechTool''']
 SCREAMING_SNAKE_CASE_    :       Dict       =							['''TranslationTool''']

if TYPE_CHECKING:
 from .agents import Agent, AzureOpenAiAgent, HfAgent, LocalAgent, OpenAiAgent
 from .base import PipelineTool, RemoteTool, Tool, launch_gradio_demo, load_tool

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .document_question_answering import DocumentQuestionAnsweringTool
  from .image_captioning import ImageCaptioningTool
  from .image_question_answering import ImageQuestionAnsweringTool
  from .image_segmentation import ImageSegmentationTool
  from .speech_to_text import SpeechToTextTool
  from .text_classification import TextClassificationTool
  from .text_question_answering import TextQuestionAnsweringTool
  from .text_summarization import TextSummarizationTool
  from .text_to_speech import TextToSpeechTool
  from .translation import TranslationTool
else:
 import sys

 SCREAMING_SNAKE_CASE_    :       Optional[Any]       =							_LazyModule(__name__, globals()['''__file__'''], _import_structure, module_spec=__spec__)






