

'''simple docstring'''


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


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

lowerCAmelCase__						:       Any										=						'''▁'''

lowerCAmelCase__						:       List[str]										=						{'''vocab_file''': '''sentencepiece.bpe.model'''}

lowerCAmelCase__						:       Tuple										=						{
    '''vocab_file''': {
        '''facebook/xglm-564M''': '''https://huggingface.co/facebook/xglm-564M/resolve/main/sentencepiece.bpe.model''',
    }
}

lowerCAmelCase__						:       Optional[int]										=						{
    '''facebook/xglm-564M''': 2048,
}




class     a			(				_UpperCAmelCase       ):






						"""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[str]			,			snake_case_					:      Any="<s>"			,			snake_case_					:      Union[str, Any]="</s>"			,			snake_case_					:      Optional[Any]="</s>"			,			snake_case_					:      List[str]="<s>"			,			snake_case_					:      Optional[int]="<unk>"			,			snake_case_					:      Union[str, Any]="<pad>"			,			snake_case_					:      Optional[Dict[str, Any]] = None			,			**snake_case_					:      List[Any]			,			):



								'''simple docstring'''



								snake_case__						:       Optional[int]       						=  {} if sp_model_kwargs is None else sp_model_kwargs

								# Compatibility with the original tokenizer
								snake_case__						:       List[str]       						=  7
								snake_case__						:       int       						=  [F"""<madeupword{i}>""" for i in range(self.num_madeup_words  )]

								snake_case__						:       Optional[Any]       						=  kwargs.get('''additional_special_tokens'''			,			[]  )
								kwargs["additional_special_tokens"] += [
								    word for word in madeup_words if word not in kwargs["additional_special_tokens"]
								]

								super().__init__(
								    bos_token=__UpperCamelCase			,			eos_token=__UpperCamelCase			,			unk_token=__UpperCamelCase			,			sep_token=__UpperCamelCase			,			cls_token=__UpperCamelCase			,			pad_token=__UpperCamelCase			,			sp_model_kwargs=self.sp_model_kwargs			,			**__UpperCamelCase			,			)

								snake_case__						:       Union[str, Any]       						=  spm.SentencePieceProcessor(**self.sp_model_kwargs  )
								self.sp_model.Load(str(__UpperCamelCase  )  )
								snake_case__						:       Union[str, Any]       						=  vocab_file

								# Original fairseq vocab and spm vocab must be "aligned":
								# Vocab    |    0    |    1    |   2    |    3    |  4  |  5  |  6  |   7   |   8   |  9
								# -------- | ------- | ------- | ------ | ------- | --- | --- | --- | ----- | ----- | ----
								# fairseq  | '<s>'   | '<pad>' | '</s>' | '<unk>' | ',' | '.' | '▁' | 's'   | '▁de' | '-'
								# spm      | '<unk>' | '<s>'   | '</s>' | ','     | '.' | '▁' | 's' | '▁de' | '-'   | '▁a'

								# The first "real" token "," has position 4 in the original fairseq vocab and position 3 in the spm vocab
								snake_case__						:       Optional[int]       						=  1

								# Mimic fairseq token-to-id alignment for the first 4 token
								snake_case__						:       Union[str, Any]       						=  {'''<s>''': 0, '''<pad>''': 1, '''</s>''': 2, '''<unk>''': 3}

								snake_case__						:       Tuple       						=  len(self.sp_model  )
								snake_case__						:       List[str]       						=  {F"""<madeupword{i}>""": sp_size + i + self.fairseq_offset for i in range(self.num_madeup_words  )}
								self.fairseq_tokens_to_ids.update(__UpperCamelCase  )

								snake_case__						:       str       						=  {v: k for k, v in self.fairseq_tokens_to_ids.items()}







						def __getstate__(							self					:      Optional[Any]  ):



								'''simple docstring'''



								snake_case__						:       Tuple       						=  self.__dict__.copy()
								snake_case__						:       Any       						=  None
								snake_case__						:       List[str]       						=  self.sp_model.serialized_model_proto()
								return state







						def __setstate__(							self					:      Dict			,			snake_case_					:      Optional[Any]  ):



								'''simple docstring'''



								snake_case__						:       List[Any]       						=  d

								# for backward compatibility
								if not hasattr(self			,			'''sp_model_kwargs'''  ):
										snake_case__						:       Optional[int]       						=  {}

								snake_case__						:       Optional[int]       						=  spm.SentencePieceProcessor(**self.sp_model_kwargs  )
								self.sp_model.LoadFromSerializedProto(self.sp_model_proto  )







						def       __magic_name__							(							self					:      Optional[int]			,			snake_case_					:      List[int]			,			snake_case_					:      Optional[List[int]] = None  ):



								'''simple docstring'''



								if token_ids_a is None:
										return [self.sep_token_id] + token_ids_a
								snake_case__						:       Optional[int]       						=  [self.sep_token_id]
								return sep + token_ids_a + sep + sep + token_ids_a







						def       __magic_name__							(							self					:      Dict			,			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=__UpperCamelCase			,			token_ids_a=__UpperCamelCase			,			already_has_special_tokens=__UpperCamelCase  )

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







						def       __magic_name__							(							self					:      Any			,			snake_case_					:      List[int]			,			snake_case_					:      Optional[List[int]] = None  ):



								'''simple docstring'''



								snake_case__						:       Tuple       						=  [self.sep_token_id]

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







						@property
						def       __magic_name__							(							self					:      Optional[Any]  ):



								'''simple docstring'''



								return len(self.sp_model  ) + self.fairseq_offset + self.num_madeup_words







						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



								snake_case__						:       str       						=  {self.convert_ids_to_tokens(__UpperCamelCase  ): i for i in range(self.vocab_size  )}
								vocab.update(self.added_tokens_encoder  )
								return vocab







						def       __magic_name__							(							self					:      Optional[int]			,			snake_case_					:      str  ):



								'''simple docstring'''



								return self.sp_model.encode(__UpperCamelCase			,			out_type=__UpperCamelCase  )







						def       __magic_name__							(							self					:      List[Any]			,			snake_case_					:      Dict  ):



								'''simple docstring'''



								if token in self.fairseq_tokens_to_ids:
										return self.fairseq_tokens_to_ids[token]
								snake_case__						:       Any       						=  self.sp_model.PieceToId(__UpperCamelCase  )

								# Need to return unknown token if the SP model returned 0
								return spm_id + self.fairseq_offset if spm_id else self.unk_token_id







						def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      Union[str, Any]  ):



								'''simple docstring'''



								if index in self.fairseq_ids_to_tokens:
										return self.fairseq_ids_to_tokens[index]
								return self.sp_model.IdToPiece(index - self.fairseq_offset  )







						def       __magic_name__							(							self					:      Optional[Any]			,			snake_case_					:      List[str]  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  ''''''.join(__UpperCamelCase  ).replace(__UpperCamelCase			,			''' '''  ).strip()
								return out_string







						def       __magic_name__							(							self					:      Any			,			snake_case_					:      str			,			snake_case_					:      Optional[str] = None  ):



								'''simple docstring'''



								if not os.path.isdir(__UpperCamelCase  ):
										logger.error(F"""Vocabulary path ({save_directory}) should be a directory"""  )
										return
								snake_case__						:       Dict       						=  os.path.join(
								    __UpperCamelCase			,			(filename_prefix + '''-''' if filename_prefix else '''''') + VOCAB_FILES_NAMES['''vocab_file''']  )

								if os.path.abspath(self.vocab_file  ) != os.path.abspath(__UpperCamelCase  ) and os.path.isfile(self.vocab_file  ):
										copyfile(self.vocab_file			,			__UpperCamelCase  )
								elif not os.path.isfile(self.vocab_file  ):
										with open(__UpperCamelCase			,			'''wb'''  ) as fi:
												snake_case__						:       str       						=  self.sp_model.serialized_model_proto()
												fi.write(__UpperCamelCase  )

								return (out_vocab_file,)






'''simple docstring'''


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

					lowerCAmelCase__						:       Any										=						True
except (ImportError, AttributeError):
					lowerCAmelCase__						:       Dict										=						object




					def 	_a       (				*__lowerCAmelCase		:      List[str]   ,							**__lowerCAmelCase		:      Tuple     ):

							"""simple docstring"""

							pass





					lowerCAmelCase__						:       str										=						False


lowerCAmelCase__						:       List[str]										=						logging.get_logger("""transformers-cli/serving""")




def 	_a       (				__lowerCAmelCase		:      Namespace     ):

		"""simple docstring"""

		snake_case__						:       Union[str, Any]       						=  pipeline(
		    task=args.task   ,							model=args.model if args.model else None   ,							config=args.config   ,							tokenizer=args.tokenizer   ,							device=args.device   ,							)
		return ServeCommand(__lowerCAmelCase   ,							args.host   ,							args.port   ,							args.workers     )




class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42
						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						@staticmethod
						def       __magic_name__							(							snake_case_					:      ArgumentParser  ):



								'''simple docstring'''



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







						def __init__(							self					:      Union[str, Any]			,			snake_case_					:      Pipeline			,			snake_case_					:      str			,			snake_case_					:      int			,			snake_case_					:      int  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  pipeline

								snake_case__						:       Tuple       						=  host
								snake_case__						:       Optional[Any]       						=  port
								snake_case__						:       Tuple       						=  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}"""  )
										snake_case__						:       str       						=  FastAPI(
										    routes=[
										        APIRoute(
										            '''/'''			,			self.model_info			,			response_model=snake_case_			,			response_class=snake_case_			,			methods=['''GET''']			,			),
										        APIRoute(
										            '''/tokenize'''			,			self.tokenize			,			response_model=snake_case_			,			response_class=snake_case_			,			methods=['''POST''']			,			),
										        APIRoute(
										            '''/detokenize'''			,			self.detokenize			,			response_model=snake_case_			,			response_class=snake_case_			,			methods=['''POST''']			,			),
										        APIRoute(
										            '''/forward'''			,			self.forward			,			response_model=snake_case_			,			response_class=snake_case_			,			methods=['''POST''']			,			),
										    ]			,			timeout=6_0_0			,			)







						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



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







						def       __magic_name__							(							self					:      Optional[Any]  ):



								'''simple docstring'''



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







						def       __magic_name__							(							self					:      List[str]			,			snake_case_					:      str = Body(snake_case_			,			embed=snake_case_  )			,			snake_case_					:      bool = Body(snake_case_			,			embed=snake_case_  )  ):



								'''simple docstring'''



								try:
										snake_case__						:       Optional[Any]       						=  self._pipeline.tokenizer.tokenize(snake_case_  )

										if return_ids:
												snake_case__						:       Optional[int]       						=  self._pipeline.tokenizer.convert_tokens_to_ids(snake_case_  )
												return ServeTokenizeResult(tokens=snake_case_			,			tokens_ids=snake_case_  )
										else:
												return ServeTokenizeResult(tokens=snake_case_  )

								except Exception as e:
										raise HTTPException(status_code=5_0_0			,			detail={'''model''': '''''', '''error''': str(snake_case_  )}  )







						def       __magic_name__							(							self					:      List[Any]			,			snake_case_					:      List[int] = Body(snake_case_			,			embed=snake_case_  )			,			snake_case_					:      bool = Body(snake_case_			,			embed=snake_case_  )			,			snake_case_					:      bool = Body(snake_case_			,			embed=snake_case_  )			,			):



								'''simple docstring'''



								try:
										snake_case__						:       Optional[int]       						=  self._pipeline.tokenizer.decode(snake_case_			,			snake_case_			,			snake_case_  )
										return ServeDeTokenizeResult(model=''''''			,			text=snake_case_  )
								except Exception as e:
										raise HTTPException(status_code=5_0_0			,			detail={'''model''': '''''', '''error''': str(snake_case_  )}  )







						async def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      List[str]=Body(snake_case_			,			embed=snake_case_  )  ):



								'''simple docstring'''



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

								try:
										# Forward through the model
										snake_case__						:       Tuple       						=  self._pipeline(snake_case_  )
										return ServeForwardResult(output=snake_case_  )
								except Exception as e:
										raise HTTPException(5_0_0			,			{'''error''': str(snake_case_  )}  )



