



"""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    : int							=							True
except (ImportError, AttributeError):
 _lowerCAmelCase    : List[str]							=							object

 def 						__snake_case						(				*SCREAMING_SNAKE_CASE__     :						Dict						,  **SCREAMING_SNAKE_CASE__     :						List[str]    )  ->       List[Any]:





      '''simple docstring'''

      pass
 _lowerCAmelCase    : Union[str, Any]							=							False


_lowerCAmelCase    : Optional[Any]							=							logging.get_logger("transformers-cli/serving")

def 						__snake_case						(				SCREAMING_SNAKE_CASE__     :						Namespace    )  ->       Optional[int]:





     '''simple docstring'''

     _UpperCAmelCase      :     List[Any]        =   pipeline(
         task=args.task						,  model=args.model if args.model else None						,  config=args.config						,  tokenizer=args.tokenizer						,  device=args.device						,  )
     return ServeCommand(SCREAMING_SNAKE_CASE__						,  args.host						,  args.port						,  args.workers    )






class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       dict







class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       List[str]
      __SCREAMING_SNAKE_CASE						:       Optional[List[int]]







class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       str







class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       Any







class      UpperCAmelCase_  (    _UpperCamelCase						):



      @staticmethod
      def 						snake_case_ (					A					:     ArgumentParser						):
           _UpperCAmelCase      :     str        =   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=8_8_8_8					,     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					:     int					,     A					:     Pipeline					,     A					:     str					,     A					:     int					,     A					:     int						):
           _UpperCAmelCase      :     Optional[int]        =   pipeline

           _UpperCAmelCase      :     List[Any]        =   host
           _UpperCAmelCase      :     List[Any]        =   port
           _UpperCAmelCase      :     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}'						)
                _UpperCAmelCase      :     List[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=6_0_0					,     )



      def 						snake_case_ (					self					:     Tuple						):
           run(self._app					,     host=self.host					,     port=self.port					,     workers=self.workers						)



      def 						snake_case_ (					self					:     int						):
           return ServeModelInfoResult(infos=vars(self._pipeline.model.config						)						)



      def 						snake_case_ (					self					:     List[str]					,     A					:     str = Body(A					,     embed=A						)					,     A					:     bool = Body(A					,     embed=A						)						):
           try:
                _UpperCAmelCase      :     Any        =   self._pipeline.tokenizer.tokenize(A						)

                if return_ids:
                     _UpperCAmelCase      :     Optional[int]        =   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=5_0_0					,     detail={"model": "", "error": str(A						)}						)



      def 						snake_case_ (					self					:     Tuple					,     A					:     List[int] = Body(A					,     embed=A						)					,     A					:     bool = Body(A					,     embed=A						)					,     A					:     bool = Body(A					,     embed=A						)					,     ):
           try:
                _UpperCAmelCase      :     List[Any]        =   self._pipeline.tokenizer.decode(A					,     A					,     A						)
                return ServeDeTokenizeResult(model=""					,     text=A						)
           except Exception as e:
                raise HTTPException(status_code=5_0_0					,     detail={"model": "", "error": str(A						)}						)



      async def 						snake_case_ (					self					:     Optional[Any]					,     A					:     Tuple=Body(A					,     embed=A						)						):

           # Check we don't have empty string
           if len(A						) == 0:
                return ServeForwardResult(output=[]					,     attention=[]						)

           try:
                # Forward through the model
                _UpperCAmelCase      :     Optional[Any]        =   self._pipeline(A						)
                return ServeForwardResult(output=A						)
           except Exception as e:
                raise HTTPException(5_0_0					,     {"error": str(A						)}						)





"""simple docstring"""




import json
from typing import List, Optional, Tuple

from tokenizers import normalizers

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from .tokenization_electra import ElectraTokenizer


_lowerCAmelCase    : List[str]							=							{"vocab_file": "vocab.txt", "tokenizer_file": "tokenizer.json"}

_lowerCAmelCase    : int							=							{
    "vocab_file": {
        "google/electra-small-generator": (
            "https://huggingface.co/google/electra-small-generator/resolve/main/vocab.txt"
        ),
        "google/electra-base-generator": "https://huggingface.co/google/electra-base-generator/resolve/main/vocab.txt",
        "google/electra-large-generator": (
            "https://huggingface.co/google/electra-large-generator/resolve/main/vocab.txt"
        ),
        "google/electra-small-discriminator": (
            "https://huggingface.co/google/electra-small-discriminator/resolve/main/vocab.txt"
        ),
        "google/electra-base-discriminator": (
            "https://huggingface.co/google/electra-base-discriminator/resolve/main/vocab.txt"
        ),
        "google/electra-large-discriminator": (
            "https://huggingface.co/google/electra-large-discriminator/resolve/main/vocab.txt"
        ),
    },
    "tokenizer_file": {
        "google/electra-small-generator": (
            "https://huggingface.co/google/electra-small-generator/resolve/main/tokenizer.json"
        ),
        "google/electra-base-generator": (
            "https://huggingface.co/google/electra-base-generator/resolve/main/tokenizer.json"
        ),
        "google/electra-large-generator": (
            "https://huggingface.co/google/electra-large-generator/resolve/main/tokenizer.json"
        ),
        "google/electra-small-discriminator": (
            "https://huggingface.co/google/electra-small-discriminator/resolve/main/tokenizer.json"
        ),
        "google/electra-base-discriminator": (
            "https://huggingface.co/google/electra-base-discriminator/resolve/main/tokenizer.json"
        ),
        "google/electra-large-discriminator": (
            "https://huggingface.co/google/electra-large-discriminator/resolve/main/tokenizer.json"
        ),
    },
}

_lowerCAmelCase    : List[Any]							=							{
    "google/electra-small-generator": 5_12,
    "google/electra-base-generator": 5_12,
    "google/electra-large-generator": 5_12,
    "google/electra-small-discriminator": 5_12,
    "google/electra-base-discriminator": 5_12,
    "google/electra-large-discriminator": 5_12,
}

_lowerCAmelCase    : Optional[Any]							=							{
    "google/electra-small-generator": {"do_lower_case": True},
    "google/electra-base-generator": {"do_lower_case": True},
    "google/electra-large-generator": {"do_lower_case": True},
    "google/electra-small-discriminator": {"do_lower_case": True},
    "google/electra-base-discriminator": {"do_lower_case": True},
    "google/electra-large-discriminator": {"do_lower_case": True},
}






class      UpperCAmelCase_  (    _UpperCamelCase						):
      __SCREAMING_SNAKE_CASE						:       Optional[Any]								=			VOCAB_FILES_NAMES
      __SCREAMING_SNAKE_CASE						:       Any								=			PRETRAINED_VOCAB_FILES_MAP
      __SCREAMING_SNAKE_CASE						:       List[str]								=			PRETRAINED_INIT_CONFIGURATION
      __SCREAMING_SNAKE_CASE						:       List[str]								=			PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
      __SCREAMING_SNAKE_CASE						:       Tuple								=			ElectraTokenizer



      def __init__(					self					:     Dict					,     A					:     Dict=None					,     A					:     Optional[int]=None					,     A					:     Dict=True					,     A					:     Optional[Any]="[UNK]"					,     A					:     Any="[SEP]"					,     A					:     str="[PAD]"					,     A					:     Tuple="[CLS]"					,     A					:     Optional[Any]="[MASK]"					,     A					:     Any=True					,     A					:     Tuple=None					,     **A					:     Any					,     ):
           super().__init__(
               A					,     tokenizer_file=A					,     do_lower_case=A					,     unk_token=A					,     sep_token=A					,     pad_token=A					,     cls_token=A					,     mask_token=A					,     tokenize_chinese_chars=A					,     strip_accents=A					,     **A					,     )

           _UpperCAmelCase      :     List[str]        =   json.loads(self.backend_tokenizer.normalizer.__getstate__()						)
           if (
               normalizer_state.get("lowercase"					,     A						) != do_lower_case
               or normalizer_state.get("strip_accents"					,     A						) != strip_accents
               or normalizer_state.get("handle_chinese_chars"					,     A						) != tokenize_chinese_chars
           ):
                _UpperCAmelCase      :     Union[str, Any]        =   getattr(A					,     normalizer_state.pop("type"						)						)
                _UpperCAmelCase      :     Dict        =   do_lower_case
                _UpperCAmelCase      :     Optional[int]        =   strip_accents
                _UpperCAmelCase      :     Any        =   tokenize_chinese_chars
                _UpperCAmelCase      :     Optional[Any]        =   normalizer_class(**A						)

           _UpperCAmelCase      :     int        =   do_lower_case



      def 						snake_case_ (					self					:     Tuple					,     A					:     str					,     A					:     int=None						):
           _UpperCAmelCase      :     Dict        =   [self.cls_token_id] + token_ids_a + [self.sep_token_id]

           if token_ids_a:
                output += token_ids_a + [self.sep_token_id]

           return output



      def 						snake_case_ (					self					:     Any					,     A					:     List[int]					,     A					:     Optional[List[int]] = None						):
           _UpperCAmelCase      :     Any        =   [self.sep_token_id]
           _UpperCAmelCase      :     Dict        =   [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 						snake_case_ (					self					:     Any					,     A					:     str					,     A					:     Optional[str] = None						):
           _UpperCAmelCase      :     List[Any]        =   self._tokenizer.model.save(A					,     name=A						)
           return tuple(A						)
