






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

     UpperCAmelCase__				   =		True
except (ImportError, AttributeError):
     UpperCAmelCase__				   =		object
     def 						_a    (      *a :str	,							**a :Optional[Any] )							->			int:
      pass

     UpperCAmelCase__				   =		False


UpperCAmelCase__				   =		logging.get_logger("transformers-cli/serving")
def 						_a    (      a :Namespace )							->			Any:
 a		  =      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 		lowercase_					(	lowercase   ):






      '''simple docstring'''




      __snake_case       =  42






class 		lowercase_					(	lowercase   ):






      '''simple docstring'''




      __snake_case       =  42
      __snake_case       =  42






class 		lowercase_					(	lowercase   ):






      '''simple docstring'''




      __snake_case       =  42






class 		lowercase_					(	lowercase   ):






      '''simple docstring'''




      __snake_case       =  42






class 		lowercase_					(	lowercase   ):






      '''simple docstring'''




      @staticmethod
      def       __lowerCAmelCase					(      __UpperCAmelCase      :						ArgumentParser  )    ->Dict:




       """simple docstring"""


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







      def __init__(      self      :						Dict ,					__UpperCAmelCase      :						Pipeline ,					__UpperCAmelCase      :						str ,					__UpperCAmelCase      :						int ,					__UpperCAmelCase      :						int  )    ->List[Any]:




       """simple docstring"""


       a		  =      pipeline

       a		  =      host
       a		  =      port
       a		  =      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}"""  )
        a		  =      FastAPI(
            routes=[
                APIRoute(
                    '''/''' ,					self.model_info ,					response_model=__UpperCAmelCase ,					response_class=__UpperCAmelCase ,					methods=['''GET'''] ,					),
                APIRoute(
                    '''/tokenize''' ,					self.tokenize ,					response_model=__UpperCAmelCase ,					response_class=__UpperCAmelCase ,					methods=['''POST'''] ,					),
                APIRoute(
                    '''/detokenize''' ,					self.detokenize ,					response_model=__UpperCAmelCase ,					response_class=__UpperCAmelCase ,					methods=['''POST'''] ,					),
                APIRoute(
                    '''/forward''' ,					self.forward ,					response_model=__UpperCAmelCase ,					response_class=__UpperCAmelCase ,					methods=['''POST'''] ,					),
            ] ,					timeout=600 ,					)







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




       """simple docstring"""


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







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




       """simple docstring"""


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







      def       __lowerCAmelCase					(      self      :						Tuple ,					__UpperCAmelCase      :						str = Body(__UpperCAmelCase ,					embed=__UpperCAmelCase  ) ,					__UpperCAmelCase      :						bool = Body(__UpperCAmelCase ,					embed=__UpperCAmelCase  )  )    ->Tuple:




       """simple docstring"""


       try:
        a		  =      self._pipeline.tokenizer.tokenize(__UpperCAmelCase  )

        if return_ids:
         a		  =      self._pipeline.tokenizer.convert_tokens_to_ids(__UpperCAmelCase  )
         return ServeTokenizeResult(tokens=__UpperCAmelCase ,					tokens_ids=__UpperCAmelCase  )
        else:
         return ServeTokenizeResult(tokens=__UpperCAmelCase  )

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







      def       __lowerCAmelCase					(      self      :						Optional[int] ,					__UpperCAmelCase      :						List[int] = Body(__UpperCAmelCase ,					embed=__UpperCAmelCase  ) ,					__UpperCAmelCase      :						bool = Body(__UpperCAmelCase ,					embed=__UpperCAmelCase  ) ,					__UpperCAmelCase      :						bool = Body(__UpperCAmelCase ,					embed=__UpperCAmelCase  ) ,					)    ->Any:




       """simple docstring"""


       try:
        a		  =      self._pipeline.tokenizer.decode(__UpperCAmelCase ,					__UpperCAmelCase ,					__UpperCAmelCase  )
        return ServeDeTokenizeResult(model='''''' ,					text=__UpperCAmelCase  )
       except Exception as e:
        raise HTTPException(status_code=500 ,					detail={'''model''': '''''', '''error''': str(__UpperCAmelCase  )}  )







      async def       __lowerCAmelCase					(      self      :						List[str] ,					__UpperCAmelCase      :						Optional[int]=Body(__UpperCAmelCase ,					embed=__UpperCAmelCase  )  )    ->Any:




       """simple docstring"""


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

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






from collections import OrderedDict
from typing import Any, Mapping, Optional

from ... import PreTrainedTokenizer
from ...configuration_utils import PretrainedConfig
from ...file_utils import TensorType, is_torch_available
from ...onnx import OnnxConfig, OnnxConfigWithPast, OnnxSeqaSeqConfigWithPast
from ...onnx.utils import compute_effective_axis_dimension
from ...utils import logging


snake_case				:    List[str]       				=     logging.get_logger(__name__)

snake_case				:    int       				=     {
    '''facebook/blenderbot_small-90M''': '''https://huggingface.co/facebook/blenderbot_small-90M/resolve/main/config.json''',
    # See all BlenderbotSmall models at https://huggingface.co/models?filter=blenderbot_small
}







class 		_snake_case  (      _snake_case							):
 SCREAMING_SNAKE_CASE__      =       'blenderbot-small'
 SCREAMING_SNAKE_CASE__      =       ['past_key_values']
 SCREAMING_SNAKE_CASE__      =       {'num_attention_heads': 'encoder_attention_heads', 'hidden_size': 'd_model'}




 def __init__(		self					,  _lowerCamelCase=5_0265					,  _lowerCamelCase=512					,  _lowerCamelCase=8					,  _lowerCamelCase=2048					,  _lowerCamelCase=16					,  _lowerCamelCase=8					,  _lowerCamelCase=2048					,  _lowerCamelCase=16					,  _lowerCamelCase=0.0					,  _lowerCamelCase=0.0					,  _lowerCamelCase=True					,  _lowerCamelCase=True					,  _lowerCamelCase="gelu"					,  _lowerCamelCase=512					,  _lowerCamelCase=0.1					,  _lowerCamelCase=0.0					,  _lowerCamelCase=0.0					,  _lowerCamelCase=0.02					,  _lowerCamelCase=1					,  _lowerCamelCase=False					,  _lowerCamelCase=0					,  _lowerCamelCase=1					,  _lowerCamelCase=2					,  _lowerCamelCase=2					,  **_lowerCamelCase					,  ):
       a						:Dict 							=       vocab_size
       a						:Optional[Any] 							=       max_position_embeddings
       a						:str 							=       d_model
       a						:Any 							=       encoder_ffn_dim
       a						:Optional[int] 							=       encoder_layers
       a						:List[str] 							=       encoder_attention_heads
       a						:List[str] 							=       decoder_ffn_dim
       a						:Optional[int] 							=       decoder_layers
       a						:str 							=       decoder_attention_heads
       a						:List[str] 							=       dropout
       a						:Optional[int] 							=       attention_dropout
       a						:Dict 							=       activation_dropout
       a						:List[str] 							=       activation_function
       a						:List[Any] 							=       init_std
       a						:Optional[int] 							=       encoder_layerdrop
       a						:Tuple 							=       decoder_layerdrop
       a						:List[str] 							=       use_cache
       a						:int 							=       encoder_layers
       a						:Union[str, Any] 							=       scale_embedding  # scale factor will be sqrt(d_model) if True

       super().__init__(
           pad_token_id=_lowerCamelCase					,  bos_token_id=_lowerCamelCase					,  eos_token_id=_lowerCamelCase					,  is_encoder_decoder=_lowerCamelCase					,  decoder_start_token_id=_lowerCamelCase					,  forced_eos_token_id=_lowerCamelCase					,  **_lowerCamelCase					,  )







class 		_snake_case  (      _snake_case							):




 @property
 def        SCREAMING_SNAKE_CASE__				(		self    ):
       if self.task in ["default", "seq2seq-lm"]:
             a						:Optional[Any] 							=       OrderedDict(
                 [
                     ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                     ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
                 ]    )

             if self.use_past:
                   a						:Union[str, Any] 							=       {0: '''batch'''}
                   a						:Tuple 							=       {0: '''batch''', 1: '''past_decoder_sequence + sequence'''}
             else:
                   a						:Optional[int] 							=       {0: '''batch''', 1: '''decoder_sequence'''}
                   a						:str 							=       {0: '''batch''', 1: '''decoder_sequence'''}

             if self.use_past:
                   self.fill_with_past_key_values_(_lowerCamelCase					,  direction='''inputs'''    )
       elif self.task == "causal-lm":
             # TODO: figure this case out.
             a						:Optional[int] 							=       OrderedDict(
                 [
                     ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                     ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
                 ]    )
             if self.use_past:
                   a							,				a						:str 							=       self.num_layers
                   for i in range(_lowerCamelCase    ):
                         a						:List[Any] 							=       {0: '''batch''', 2: '''past_sequence + sequence'''}
                         a						:List[str] 							=       {0: '''batch''', 2: '''past_sequence + sequence'''}
       else:
             a						:Optional[int] 							=       OrderedDict(
                 [
                     ('''input_ids''', {0: '''batch''', 1: '''encoder_sequence'''}),
                     ('''attention_mask''', {0: '''batch''', 1: '''encoder_sequence'''}),
                     ('''decoder_input_ids''', {0: '''batch''', 1: '''decoder_sequence'''}),
                     ('''decoder_attention_mask''', {0: '''batch''', 1: '''decoder_sequence'''}),
                 ]    )

       return common_inputs




 @property
 def        SCREAMING_SNAKE_CASE__				(		self    ):
       if self.task in ["default", "seq2seq-lm"]:
             a						:List[Any] 							=       super().outputs
       else:
             a						:Union[str, Any] 							=       super(_lowerCamelCase					,  self    ).outputs
             if self.use_past:
                   a							,				a						:int 							=       self.num_layers
                   for i in range(_lowerCamelCase    ):
                         a						:int 							=       {0: '''batch''', 2: '''past_sequence + sequence'''}
                         a						:Optional[Any] 							=       {0: '''batch''', 2: '''past_sequence + sequence'''}
       return common_outputs




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = -1					,  _lowerCamelCase = -1					,  _lowerCamelCase = False					,  _lowerCamelCase = None					,  ):
       a						:Tuple 							=       self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
           _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    )

       # Generate decoder inputs
       a						:Dict 							=       seq_length if not self.use_past else 1
       a						:Dict 							=       self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
           _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    )
       a						:List[Any] 							=       {F'''decoder_{name}''': tensor for name, tensor in decoder_inputs.items()}
       a						:List[str] 							=       dict(**_lowerCamelCase					,  **_lowerCamelCase    )

       if self.use_past:
             if not is_torch_available():
                   raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''    )
             else:
                   import torch
             a							,				a						:Optional[Any] 							=       common_inputs['''input_ids'''].shape
             a						:Tuple 							=       common_inputs['''decoder_input_ids'''].shape[1]
             a							,				a						:List[Any] 							=       self.num_attention_heads
             a						:List[Any] 							=       (
                 batch,
                 num_encoder_attention_heads,
                 encoder_seq_length,
                 self._config.hidden_size // num_encoder_attention_heads,
             )
             a						:int 							=       decoder_seq_length + 3
             a						:Union[str, Any] 							=       (
                 batch,
                 num_decoder_attention_heads,
                 decoder_past_length,
                 self._config.hidden_size // num_decoder_attention_heads,
             )

             a						:Union[str, Any] 							=       torch.cat(
                 [common_inputs['''decoder_attention_mask'''], torch.ones(_lowerCamelCase					,  _lowerCamelCase    )]					,  dim=1    )

             a						:List[Any] 							=       []
             # If the number of encoder and decoder layers are present in the model configuration, both are considered
             a							,				a						:Optional[int] 							=       self.num_layers
             a						:str 							=       min(_lowerCamelCase					,  _lowerCamelCase    )
             a						:str 							=       max(_lowerCamelCase					,  _lowerCamelCase    ) - min_num_layers
             a						:Tuple 							=       '''encoder''' if num_encoder_layers > num_decoder_layers else '''decoder'''

             for _ in range(_lowerCamelCase    ):
                   common_inputs["past_key_values"].append(
                       (
                           torch.zeros(_lowerCamelCase    ),
                           torch.zeros(_lowerCamelCase    ),
                           torch.zeros(_lowerCamelCase    ),
                           torch.zeros(_lowerCamelCase    ),
                       )    )
             # TODO: test this.
             a						:int 							=       encoder_shape if remaining_side_name == '''encoder''' else decoder_shape
             for _ in range(_lowerCamelCase					,  _lowerCamelCase    ):
                   common_inputs["past_key_values"].append((torch.zeros(_lowerCamelCase    ), torch.zeros(_lowerCamelCase    ))    )
       return common_inputs




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = -1					,  _lowerCamelCase = -1					,  _lowerCamelCase = False					,  _lowerCamelCase = None					,  ):
       a						:Tuple 							=       self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
           _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    )

       if self.use_past:
             if not is_torch_available():
                   raise ValueError('''Cannot generate dummy past_keys inputs without PyTorch installed.'''    )
             else:
                   import torch
             a							,				a						:Dict 							=       common_inputs['''input_ids'''].shape
             # Not using the same length for past_key_values
             a						:Optional[int] 							=       seqlen + 2
             a							,				a						:Union[str, Any] 							=       self.num_layers
             a							,				a						:Optional[Any] 							=       self.num_attention_heads
             a						:str 							=       (
                 batch,
                 num_encoder_attention_heads,
                 past_key_values_length,
                 self._config.hidden_size // num_encoder_attention_heads,
             )

             a						:Tuple 							=       common_inputs['''attention_mask'''].dtype
             a						:Any 							=       torch.cat(
                 [common_inputs['''attention_mask'''], torch.ones(_lowerCamelCase					,  _lowerCamelCase					,  dtype=_lowerCamelCase    )]					,  dim=1    )
             a						:Any 							=       [
                 (torch.zeros(_lowerCamelCase    ), torch.zeros(_lowerCamelCase    )) for _ in range(_lowerCamelCase    )
             ]
       return common_inputs




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = -1					,  _lowerCamelCase = -1					,  _lowerCamelCase = False					,  _lowerCamelCase = None					,  ):
       # Copied from OnnxConfig.generate_dummy_inputs
       # Did not use super(OnnxConfigWithPast, self).generate_dummy_inputs for code clarity.
       # If dynamic axis (-1) we forward with a fixed dimension of 2 samples to avoid optimizations made by ONNX
       a						:Optional[Any] 							=       compute_effective_axis_dimension(
           _lowerCamelCase					,  fixed_dimension=OnnxConfig.default_fixed_batch					,  num_token_to_add=0    )

       # If dynamic axis (-1) we forward with a fixed dimension of 8 tokens to avoid optimizations made by ONNX
       a						:Optional[int] 							=       tokenizer.num_special_tokens_to_add(_lowerCamelCase    )
       a						:Tuple 							=       compute_effective_axis_dimension(
           _lowerCamelCase					,  fixed_dimension=OnnxConfig.default_fixed_sequence					,  num_token_to_add=_lowerCamelCase    )

       # Generate dummy inputs according to compute batch and sequence
       a						:List[str] 							=       [''' '''.join([tokenizer.unk_token]    ) * seq_length] * batch_size
       a						:Dict 							=       dict(tokenizer(_lowerCamelCase					,  return_tensors=_lowerCamelCase    )    )
       return common_inputs




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase = -1					,  _lowerCamelCase = -1					,  _lowerCamelCase = False					,  _lowerCamelCase = None					,  ):
       if self.task in ["default", "seq2seq-lm"]:
             a						:Tuple 							=       self._generate_dummy_inputs_for_default_and_seqaseq_lm(
                 _lowerCamelCase					,  batch_size=_lowerCamelCase					,  seq_length=_lowerCamelCase					,  is_pair=_lowerCamelCase					,  framework=_lowerCamelCase    )

       elif self.task == "causal-lm":
             a						:Dict 							=       self._generate_dummy_inputs_for_causal_lm(
                 _lowerCamelCase					,  batch_size=_lowerCamelCase					,  seq_length=_lowerCamelCase					,  is_pair=_lowerCamelCase					,  framework=_lowerCamelCase    )
       else:
             a						:Dict 							=       self._generate_dummy_inputs_for_sequence_classification_and_question_answering(
                 _lowerCamelCase					,  batch_size=_lowerCamelCase					,  seq_length=_lowerCamelCase					,  is_pair=_lowerCamelCase					,  framework=_lowerCamelCase    )

       return common_inputs




 def        SCREAMING_SNAKE_CASE__				(		self					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    ):
       if self.task in ["default", "seq2seq-lm"]:
             a						:Optional[int] 							=       super()._flatten_past_key_values_(_lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    )
       else:
             a						:Any 							=       super(_lowerCamelCase					,  self    )._flatten_past_key_values_(
                 _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase					,  _lowerCamelCase    )





