

import json
from typing import List, Optional, Tuple

from tokenizers import normalizers
from tokenizers.pre_tokenizers import BertPreTokenizer, PreTokenizer

from ...tokenization_utils_fast import PreTrainedTokenizerFast
from ...utils import logging
from .tokenization_roformer import RoFormerTokenizer
from .tokenization_utils import JiebaPreTokenizer


lowercase			:				List[Any]   =     logging.get_logger(__name__)

lowercase			:				int   =     {"vocab_file": "vocab.txt", "tokenizer_file": "tokenizer.json"}

lowercase			:				Dict   =     {
    "vocab_file": {
        "junnyu/roformer_chinese_small": "https://huggingface.co/junnyu/roformer_chinese_small/resolve/main/vocab.txt",
        "junnyu/roformer_chinese_base": "https://huggingface.co/junnyu/roformer_chinese_base/resolve/main/vocab.txt",
        "junnyu/roformer_chinese_char_small": (
            "https://huggingface.co/junnyu/roformer_chinese_char_small/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_chinese_char_base": (
            "https://huggingface.co/junnyu/roformer_chinese_char_base/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_small_discriminator": (
            "https://huggingface.co/junnyu/roformer_small_discriminator/resolve/main/vocab.txt"
        ),
        "junnyu/roformer_small_generator": (
            "https://huggingface.co/junnyu/roformer_small_generator/resolve/main/vocab.txt"
        ),
    }
}

lowercase			:				int   =     {
    "junnyu/roformer_chinese_small": 15_36,
    "junnyu/roformer_chinese_base": 15_36,
    "junnyu/roformer_chinese_char_small": 5_12,
    "junnyu/roformer_chinese_char_base": 5_12,
    "junnyu/roformer_small_discriminator": 1_28,
    "junnyu/roformer_small_generator": 1_28,
}


lowercase			:				str   =     {
    "junnyu/roformer_chinese_small": {"do_lower_case": True},
    "junnyu/roformer_chinese_base": {"do_lower_case": True},
    "junnyu/roformer_chinese_char_small": {"do_lower_case": True},
    "junnyu/roformer_chinese_char_base": {"do_lower_case": True},
    "junnyu/roformer_small_discriminator": {"do_lower_case": True},
    "junnyu/roformer_small_generator": {"do_lower_case": True},
}







class 					__lowercase  (						SCREAMING_SNAKE_CASE_				):





							"""simple docstring"""

							UpperCAmelCase_						:		Union[str, Any]								=	VOCAB_FILES_NAMES
							UpperCAmelCase_						:		Any								=	PRETRAINED_VOCAB_FILES_MAP
							UpperCAmelCase_						:		List[str]								=	PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							UpperCAmelCase_						:		List[str]								=	PRETRAINED_INIT_CONFIGURATION
							UpperCAmelCase_						:		Any								=	RoFormerTokenizer




							def __init__(					self    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase=True    ,						__UpperCAmelCase="[UNK]"    ,						__UpperCAmelCase="[SEP]"    ,						__UpperCAmelCase="[PAD]"    ,						__UpperCAmelCase="[CLS]"    ,						__UpperCAmelCase="[MASK]"    ,						__UpperCAmelCase=True    ,						__UpperCAmelCase=None    ,						**__UpperCAmelCase    ,						)		->       Optional[int]:
														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    ,						)

														A      :					List[str]      		=					json.loads(self.backend_tokenizer.normalizer.__getstate__()						)
														if (
														    pre_tok_state.get('''lowercase'''    ,						_a						) != do_lower_case
														    or pre_tok_state.get('''strip_accents'''    ,						_a						) != strip_accents
														):
																					A      :					Dict      		=					getattr(_a    ,						pre_tok_state.pop('''type'''						)						)
																					A      :					Tuple      		=					do_lower_case
																					A      :					int      		=					strip_accents
																					A      :					Dict      		=					pre_tok_class(**_a						)

														A      :					Any      		=					do_lower_case




							def __getstate__(					self						)		->       List[Any]:
														A      :					Optional[Any]      		=					self.__dict__.copy()
														A      :					Any      		=					BertPreTokenizer()
														return state




							def __setstate__(					self    ,						__UpperCAmelCase						)		->       Tuple:
														A      :					Tuple      		=					d
														A      :					Optional[Any]      		=					self.__dict__['_tokenizer'].get_vocab()
														A      :					Any      		=					PreTokenizer.custom(JiebaPreTokenizer(_a						)						)




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






							def       snake_case			(					self    ,						__UpperCAmelCase    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase=None    ,						__UpperCAmelCase=False    ,						**__UpperCAmelCase    ,						)		->       Dict:
														A      :					Dict      		=					BertPreTokenizer()
														return super().save_pretrained(_a    ,						_a    ,						_a    ,						_a    ,						**_a						)











'''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

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

   def 			a_	(				*_lowerCAmelCase			,**_lowerCAmelCase			) -> Union[str, Any]:
     pass


   _UpperCamelCase											=		False


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

def 			a_	(				_lowerCAmelCase			) -> Optional[int]:
  __lowerCamelCase			:			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(_lowerCAmelCase			,args.host			,args.port			,args.workers			)






class    lowerCamelCase_       (							SCREAMING_SNAKE_CASE_	):
  """simple docstring"""




  a_				   =42






class    lowerCamelCase_       (							SCREAMING_SNAKE_CASE_	):
  """simple docstring"""




  a_				   =42
  a_				   =42






class    lowerCamelCase_       (							SCREAMING_SNAKE_CASE_	):
  """simple docstring"""




  a_				   =42






class    lowerCamelCase_       (							SCREAMING_SNAKE_CASE_	):
  """simple docstring"""




  a_				   =42






class    lowerCamelCase_       (							SCREAMING_SNAKE_CASE_	):
  """simple docstring"""




  @staticmethod
  def 		_lowercase	(					_a :      ArgumentParser  )   ->   Optional[Any]:
    __lowerCamelCase			:			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 :      int     ,      _a :      Pipeline     ,      _a :      str     ,      _a :      int     ,      _a :      int  )   ->   Optional[int]:
    __lowerCamelCase			:			int       			=					pipeline

    __lowerCamelCase			:			List[Any]       			=					host
    __lowerCamelCase			:			int       			=					port
    __lowerCamelCase			:			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}'  )
      __lowerCamelCase			:			Optional[Any]       			=					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 		_lowercase	(					self :      Optional[Any]  )   ->   List[str]:
    run(self._app     ,      host=self.host     ,      port=self.port     ,      workers=self.workers  )




  def 		_lowercase	(					self :      Tuple  )   ->   int:
    return ServeModelInfoResult(infos=vars(self._pipeline.model.config  )  )




  def 		_lowercase	(					self :      str     ,      _a :      str = Body(_a     ,      embed=_a  )     ,      _a :      bool = Body(_a     ,      embed=_a  )  )   ->   Optional[int]:
    try:
      __lowerCamelCase			:			Dict       			=					self._pipeline.tokenizer.tokenize(_a  )

      if return_ids:
        __lowerCamelCase			:			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=500     ,      detail={'model': '', 'error': str(_a  )}  )




  def 		_lowercase	(					self :      Optional[Any]     ,      _a :      List[int] = Body(_a     ,      embed=_a  )     ,      _a :      bool = Body(_a     ,      embed=_a  )     ,      _a :      bool = Body(_a     ,      embed=_a  )     ,      )   ->   List[Any]:
    try:
      __lowerCamelCase			:			Dict       			=					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 		_lowercase	(					self :      Dict     ,      _a :      Any=Body(_a     ,      embed=_a  )  )   ->   Optional[int]:

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

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






