





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

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






    def lowerCamelCase__   (							*snake_case_     :  int	,				**snake_case_     :  Optional[Any]	)				->    Dict:
       pass







    snake_case_           =      False


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






def lowerCamelCase__   (							snake_case_     :  Namespace	)				->    Dict:
   __snake_case								=       pipeline(
       task=args.task	,				model=args.model if args.model else None	,				config=args.config	,				tokenizer=args.tokenizer	,				device=args.device	,				)
   return ServeCommand(snake_case_	,				args.host	,				args.port	,				args.workers	)

class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):
      A_						:    dict






class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):
      A_						:    List[str]
      A_						:    Optional[List[int]]






class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):
      A_						:    str






class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):
      A_						:    Any






class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):


      @staticmethod
      def        a			(a__	:				ArgumentParser				):





         """simple docstring"""



         __snake_case								=       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	:				Any    ,      a__	:				Pipeline    ,      a__	:				str    ,      a__	:				int    ,      a__	:				int				):





         """simple docstring"""



         __snake_case								=       pipeline

         __snake_case								=       host
         __snake_case								=       port
         __snake_case								=       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								=       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        a			(self	:				Optional[int]				):





         """simple docstring"""



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


      def        a			(self	:				List[Any]				):





         """simple docstring"""



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


      def        a			(self	:				List[str]    ,      a__	:				str = Body(a__    ,      embed=a__				)    ,      a__	:				bool = Body(a__    ,      embed=a__				)				):





         """simple docstring"""



         try:
            __snake_case								=       self._pipeline.tokenizer.tokenize(a__				)

            if return_ids:
               __snake_case								=       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        a			(self	:				Union[str, Any]    ,      a__	:				List[int] = Body(a__    ,      embed=a__				)    ,      a__	:				bool = Body(a__    ,      embed=a__				)    ,      a__	:				bool = Body(a__    ,      embed=a__				)    ,      ):





         """simple docstring"""



         try:
            __snake_case								=       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        a			(self	:				Union[str, Any]    ,      a__	:				Any=Body(a__    ,      embed=a__				)				):





         """simple docstring"""



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

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








from typing import List, Optional, Tuple

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_herbert import HerbertTokenizer


snake_case_           =      logging.get_logger(__name__)

snake_case_           =      {'vocab_file': 'vocab.json', 'merges_file': 'merges.txt', 'tokenizer_file': 'tokenizer.json'}

snake_case_           =      {
    'vocab_file': {
        'allegro/herbert-base-cased': 'https://huggingface.co/allegro/herbert-base-cased/resolve/main/vocab.json'
    },
    'merges_file': {
        'allegro/herbert-base-cased': 'https://huggingface.co/allegro/herbert-base-cased/resolve/main/merges.txt'
    },
}

snake_case_           =      {'allegro/herbert-base-cased': 514}
snake_case_           =      {}

class 		SCREAMING_SNAKE_CASE__    (	_UpperCAmelCase  ):
      A_						:    Dict							=   VOCAB_FILES_NAMES
      A_						:    Optional[Any]							=   PRETRAINED_VOCAB_FILES_MAP
      A_						:    Optional[int]							=   PRETRAINED_INIT_CONFIGURATION
      A_						:    int							=   PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      A_						:    Any							=   HerbertTokenizer


      def __init__(self	:				Dict    ,      a__	:				Tuple=None    ,      a__	:				Optional[int]=None    ,      a__	:				List[str]=None    ,      a__	:				Optional[int]="<s>"    ,      a__	:				Optional[Any]="<unk>"    ,      a__	:				Any="<pad>"    ,      a__	:				List[Any]="<mask>"    ,      a__	:				Any="</s>"    ,      **a__	:				Tuple    ,      ):





         """simple docstring"""



         super().__init__(
             a__    ,      a__    ,      tokenizer_file=a__    ,      cls_token=a__    ,      unk_token=a__    ,      pad_token=a__    ,      mask_token=a__    ,      sep_token=a__    ,      **a__    ,      )


      def        a			(self	:				List[str]    ,      a__	:				List[int]    ,      a__	:				Optional[List[int]] = None				):





         """simple docstring"""



         __snake_case								=       [self.cls_token_id]
         __snake_case								=       [self.sep_token_id]
         if token_ids_a is None:
            return cls + token_ids_a + sep

         return cls + token_ids_a + sep + token_ids_a + sep


      def        a			(self	:				List[str]    ,      a__	:				List[int]    ,      a__	:				Optional[List[int]] = None    ,      a__	:				bool = False				):





         """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] + ([0] * len(a__				)) + [1]


      def        a			(self	:				Optional[int]    ,      a__	:				List[int]    ,      a__	:				Optional[List[int]] = None				):





         """simple docstring"""



         __snake_case								=       [self.sep_token_id]
         __snake_case								=       [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				) * [0] + len(token_ids_a + sep				) * [1]


      def        a			(self	:				int    ,      a__	:				str    ,      a__	:				Optional[str] = None				):





         """simple docstring"""



         __snake_case								=       self._tokenizer.model.save(a__    ,      name=a__				)
         return tuple(a__				)

