






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

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


SCREAMING_SNAKE_CASE 							=						logging.get_logger(__name__)


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

# See all LED models at https://huggingface.co/models?filter=LED
SCREAMING_SNAKE_CASE 							=						{
    'vocab_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json',
    },
    'merges_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt',
    },
    'tokenizer_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json',
    },
}

SCREAMING_SNAKE_CASE 							=						{
    'allenai/led-base-16384': 16384,
}

@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def      _lowerCamelCase	(						)      ->					Union[str, Any]:
    _UpperCAmelCase				:      Union[str, Any]			  =				(
        list(range(ord('''!''' )		,		ord('''~''' ) + 1 ) ) + list(range(ord('''¡''' )		,		ord('''¬''' ) + 1 ) ) + list(range(ord('''®''' )		,		ord('''ÿ''' ) + 1 ) )
    )
    _UpperCAmelCase				:      List[str]			  =				bs[:]
    _UpperCAmelCase				:      str			  =				0
    for b in range(2**8 ):
        if b not in bs:
            bs.append(lowercase_ )
            cs.append(2**8 + n )
            n += 1
    _UpperCAmelCase				:      str			  =				[chr(lowercase_ ) for n in cs]
    return dict(zip(lowercase_		,		lowercase_ ) )

def      _lowerCamelCase	(						__A    :							Dict )      ->					str:
    _UpperCAmelCase				:      Dict			  =				set()
    _UpperCAmelCase				:      Optional[Any]			  =				word[0]
    for char in word[1:]:
        pairs.add((prev_char, char) )
        _UpperCAmelCase				:      Dict			  =				char
    return pairs


class 							A_    (						_UpperCamelCase			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			Optional[int] =					VOCAB_FILES_NAMES
 _SCREAMING_SNAKE_CASE							:			Union[str, Any] =					PRETRAINED_VOCAB_FILES_MAP
 _SCREAMING_SNAKE_CASE							:			Tuple =					PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 _SCREAMING_SNAKE_CASE							:			Optional[int] =					["input_ids", "attention_mask"]


 def __init__(	self							,  _A							,  _A							,  _A="replace"							,  _A="<s>"							,  _A="</s>"							,  _A="</s>"							,  _A="<s>"							,  _A="<unk>"							,  _A="<pad>"							,  _A="<mask>"							,  _A=False							,  **_A							,  ) ->		str:

     """simple docstring"""



     _UpperCAmelCase				:      Tuple			  =				AddedToken(__a							,  lstrip=__a							,  rstrip=__a) if isinstance(__a							,  __a) else bos_token
     _UpperCAmelCase				:      List[str]			  =				AddedToken(__a							,  lstrip=__a							,  rstrip=__a) if isinstance(__a							,  __a) else eos_token
     _UpperCAmelCase				:      Tuple			  =				AddedToken(__a							,  lstrip=__a							,  rstrip=__a) if isinstance(__a							,  __a) else sep_token
     _UpperCAmelCase				:      str			  =				AddedToken(__a							,  lstrip=__a							,  rstrip=__a) if isinstance(__a							,  __a) else cls_token
     _UpperCAmelCase				:      Any			  =				AddedToken(__a							,  lstrip=__a							,  rstrip=__a) if isinstance(__a							,  __a) else unk_token
     _UpperCAmelCase				:      List[Any]			  =				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
     _UpperCAmelCase				:      int			  =				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:
         _UpperCAmelCase				:      int			  =				json.load(__a)
     _UpperCAmelCase				:      str			  =				{v: k for k, v in self.encoder.items()}
     _UpperCAmelCase				:      int			  =				errors  # how to handle errors in decoding
     _UpperCAmelCase				:      str			  =				bytes_to_unicode()
     _UpperCAmelCase				:      Optional[Any]			  =				{v: k for k, v in self.byte_encoder.items()}
     with open(__a							,  encoding='''utf-8''') as merges_handle:
         _UpperCAmelCase				:      List[Any]			  =				merges_handle.read().split('''\n''')[1:-1]
     _UpperCAmelCase				:      Optional[int]			  =				[tuple(merge.split()) for merge in bpe_merges]
     _UpperCAmelCase				:      List[Any]			  =				dict(zip(__a							,  range(len(__a))))
     _UpperCAmelCase				:      Any			  =				{}
     _UpperCAmelCase				:      List[Any]			  =				add_prefix_space

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


 @property
 # Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
 def  snake_case__				(	self) ->		Optional[Any]:

     """simple docstring"""



     return len(self.encoder)


 def  snake_case__				(	self) ->		Tuple:

     """simple docstring"""



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


 def  snake_case__				(	self							,  _A) ->		Tuple:

     """simple docstring"""



     if token in self.cache:
         return self.cache[token]
     _UpperCAmelCase				:      Optional[int]			  =				tuple(__a)
     _UpperCAmelCase				:      List[Any]			  =				get_pairs(__a)

     if not pairs:
         return token

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


 def  snake_case__				(	self							,  _A) ->		Tuple:

     """simple docstring"""



     _UpperCAmelCase				:      str			  =				[]
     for token in re.findall(self.pat							,  __a):
         _UpperCAmelCase				:      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(__a).split(''' '''))
     return bpe_tokens


 def  snake_case__				(	self							,  _A) ->		Tuple:

     """simple docstring"""



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


 def  snake_case__				(	self							,  _A) ->		Union[str, Any]:

     """simple docstring"""



     return self.decoder.get(__a)


 def  snake_case__				(	self							,  _A) ->		str:

     """simple docstring"""



     _UpperCAmelCase				:      Tuple			  =				"".join(__a)
     _UpperCAmelCase				:      List[Any]			  =				bytearray([self.byte_decoder[c] for c in text]).decode('''utf-8'''							,  errors=self.errors)
     return text


 def  snake_case__				(	self							,  _A							,  _A = None) ->		Tuple[str]:

     """simple docstring"""



     if not os.path.isdir(__a):
         logger.error(f'''Vocabulary path ({save_directory}) should be a directory''')
         return
     _UpperCAmelCase				:      Tuple			  =				os.path.join(
         __a							,  (filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file'''])
     _UpperCAmelCase				:      Dict			  =				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''')

     _UpperCAmelCase				:      str			  =				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!''')
                 _UpperCAmelCase				:      Union[str, Any]			  =				token_index
             writer.write(''' '''.join(__a) + '''\n''')
             index += 1

     return vocab_file, merge_file


 def  snake_case__				(	self							,  _A							,  _A = None) ->		List[int]:

     """simple docstring"""



     if token_ids_a is None:
         return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
     _UpperCAmelCase				:      Tuple			  =				[self.cls_token_id]
     _UpperCAmelCase				:      List[Any]			  =				[self.sep_token_id]
     return cls + token_ids_a + sep + sep + token_ids_a + sep


 def  snake_case__				(	self							,  _A							,  _A = None							,  _A = 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  snake_case__				(	self							,  _A							,  _A = None) ->		List[int]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[int]			  =				[self.sep_token_id]
     _UpperCAmelCase				:      List[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  snake_case__				(	self							,  _A							,  _A=False							,  **_A) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Optional[Any]			  =				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()):
         _UpperCAmelCase				:      List[Any]			  =				" " + text
     return (text, kwargs)


 def  snake_case__				(	self							,  _A							,  _A = None							,  _A = PaddingStrategy.DO_NOT_PAD							,  _A = None							,  _A = None							,  ) ->		dict:

     """simple docstring"""



     _UpperCAmelCase				:      Any			  =				super()._pad(
         encoded_inputs=__a							,  max_length=__a							,  padding_strategy=__a							,  pad_to_multiple_of=__a							,  return_attention_mask=__a							,  )

     # Load from model defaults
     if return_attention_mask is None:
         _UpperCAmelCase				:      str			  =				"attention_mask" in self.model_input_names

     if return_attention_mask and "global_attention_mask" in encoded_inputs:
         _UpperCAmelCase				:      Dict			  =				encoded_inputs[self.model_input_names[0]]
         # `global_attention_mask` need to have the same length as other (sequential) inputs.
         _UpperCAmelCase				:      str			  =				len(encoded_inputs['''global_attention_mask''']) != len(__a)

         if needs_to_be_padded:
             _UpperCAmelCase				:      List[str]			  =				len(__a) - len(encoded_inputs['''global_attention_mask'''])

             if self.padding_side == "right":
                 # Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
                 _UpperCAmelCase				:      str			  =				(
                     encoded_inputs["global_attention_mask"] + [-1] * difference
                 )
             elif self.padding_side == "left":
                 _UpperCAmelCase				:      List[Any]			  =				[-1] * difference + encoded_inputs[
                     "global_attention_mask"
                 ]
             else:
                 raise ValueError('''Invalid padding strategy:''' + str(self.padding_side))

     return encoded_inputs








from argparse import ArgumentParser, Namespace
from typing import Any, List, Optional

from ..pipelines import Pipeline, get_supported_tasks, pipeline
from ..utils import logging
from . import BaseTransformersCLICommand


try:
  from fastapi import Body, FastAPI, HTTPException
  from fastapi.routing import APIRoute
  from pydantic import BaseModel
  from starlette.responses import JSONResponse
  from uvicorn import run

  SCREAMING_SNAKE_CASE 							=						True
except (ImportError, AttributeError):
  SCREAMING_SNAKE_CASE 							=						object

  def      _lowerCamelCase	(						*__A    :							List[Any]		,		**__A    :							Any )      ->					Any:
      pass


  SCREAMING_SNAKE_CASE 							=						False


SCREAMING_SNAKE_CASE 							=						logging.get_logger('transformers-cli/serving')

def      _lowerCamelCase	(						__A    :							Namespace )      ->					Tuple:
    _UpperCAmelCase				:      Any			  =				pipeline(
        task=args.task		,		model=args.model if args.model else None		,		config=args.config		,		tokenizer=args.tokenizer		,		device=args.device		,		)
    return ServeCommand(__A		,		args.host		,		args.port		,		args.workers )
class 							A_    (						__lowercase			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			dict



class 							A_    (						__lowercase			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			List[str]
 _SCREAMING_SNAKE_CASE							:			Optional[List[int]]



class 							A_    (						__lowercase			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			str



class 							A_    (						__lowercase			):






 '''simple docstring'''




 _SCREAMING_SNAKE_CASE							:			Any



class 							A_    (						__lowercase			):






 '''simple docstring'''




 @staticmethod
 def  snake_case__				(	_A) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      Union[str, Any]			  =				parser.add_parser(
         '''serve'''							,  help='''CLI tool to run inference requests through REST and GraphQL endpoints.''')
     serve_parser.add_argument(
         '''--task'''							,  type=_A							,  choices=get_supported_tasks()							,  help='''The task to run the pipeline on'''							,  )
     serve_parser.add_argument('''--host'''							,  type=_A							,  default='''localhost'''							,  help='''Interface the server will listen on.''')
     serve_parser.add_argument('''--port'''							,  type=_A							,  default=8888							,  help='''Port the serving will listen to.''')
     serve_parser.add_argument('''--workers'''							,  type=_A							,  default=1							,  help='''Number of http workers''')
     serve_parser.add_argument('''--model'''							,  type=_A							,  help='''Model\'s name or path to stored model.''')
     serve_parser.add_argument('''--config'''							,  type=_A							,  help='''Model\'s config name or path to stored model.''')
     serve_parser.add_argument('''--tokenizer'''							,  type=_A							,  help='''Tokenizer name to use.''')
     serve_parser.add_argument(
         '''--device'''							,  type=_A							,  default=-1							,  help='''Indicate the device to run onto, -1 indicates CPU, >= 0 indicates GPU (default: -1)'''							,  )
     serve_parser.set_defaults(func=_A)


 def __init__(	self							,  _A							,  _A							,  _A							,  _A) ->		List[Any]:

     """simple docstring"""



     _UpperCAmelCase				:      List[Any]			  =				pipeline

     _UpperCAmelCase				:      List[Any]			  =				host
     _UpperCAmelCase				:      Tuple			  =				port
     _UpperCAmelCase				:      Union[str, Any]			  =				workers

     if not _serve_dependencies_installed:
         raise RuntimeError(
             '''Using serve command requires FastAPI and uvicorn. '''
             '''Please install transformers with [serving]: pip install "transformers[serving]".'''
             '''Or install FastAPI and uvicorn separately.''')
     else:
         logger.info(f'''Serving model over {host}:{port}''')
         _UpperCAmelCase				:      str			  =				FastAPI(
             routes=[
                 APIRoute(
                     '''/'''							,  self.model_info							,  response_model=_A							,  response_class=_A							,  methods=['''GET''']							,  ),
                 APIRoute(
                     '''/tokenize'''							,  self.tokenize							,  response_model=_A							,  response_class=_A							,  methods=['''POST''']							,  ),
                 APIRoute(
                     '''/detokenize'''							,  self.detokenize							,  response_model=_A							,  response_class=_A							,  methods=['''POST''']							,  ),
                 APIRoute(
                     '''/forward'''							,  self.forward							,  response_model=_A							,  response_class=_A							,  methods=['''POST''']							,  ),
             ]							,  timeout=600							,  )


 def  snake_case__				(	self) ->		Optional[Any]:

     """simple docstring"""



     run(self._app							,  host=self.host							,  port=self.port							,  workers=self.workers)


 def  snake_case__				(	self) ->		Union[str, Any]:

     """simple docstring"""



     return ServeModelInfoResult(infos=vars(self._pipeline.model.config))


 def  snake_case__				(	self							,  _A = Body(_A							,  embed=_A)							,  _A = Body(_A							,  embed=_A)) ->		List[Any]:

     """simple docstring"""



     try:
         _UpperCAmelCase				:      int			  =				self._pipeline.tokenizer.tokenize(_A)

         if return_ids:
             _UpperCAmelCase				:      List[Any]			  =				self._pipeline.tokenizer.convert_tokens_to_ids(_A)
             return ServeTokenizeResult(tokens=_A							,  tokens_ids=_A)
         else:
             return ServeTokenizeResult(tokens=_A)

     except Exception as e:
         raise HTTPException(status_code=500							,  detail={'''model''': '''''', '''error''': str(_A)})


 def  snake_case__				(	self							,  _A = Body(_A							,  embed=_A)							,  _A = Body(_A							,  embed=_A)							,  _A = Body(_A							,  embed=_A)							,  ) ->		str:

     """simple docstring"""



     try:
         _UpperCAmelCase				:      Optional[Any]			  =				self._pipeline.tokenizer.decode(_A							,  _A							,  _A)
         return ServeDeTokenizeResult(model=''''''							,  text=_A)
     except Exception as e:
         raise HTTPException(status_code=500							,  detail={'''model''': '''''', '''error''': str(_A)})


 async def  snake_case__				(	self							,  _A=Body(_A							,  embed=_A)) ->		str:

     """simple docstring"""



     if len(_A) == 0:
         return ServeForwardResult(output=[]							,  attention=[])

     try:
         # Forward through the model
         _UpperCAmelCase				:      Any			  =				self._pipeline(_A)
         return ServeForwardResult(output=_A)
     except Exception as e:
         raise HTTPException(500							,  {'''error''': str(_A)})




