


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

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


 def 					lowercase      (    *A_  ,       **A_	)->				int:
   '''simple docstring'''







   pass







 __lowercase    =				False


__lowercase    =				logging.get_logger("""transformers-cli/serving""")


def 					lowercase      (    A_	)->				Optional[int]:
  '''simple docstring'''







  a						:		str											=      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 						_A  (  _a  ):







    """simple docstring"""




    UpperCAmelCase	:							dict






class 						_A  (  _a  ):







    """simple docstring"""




    UpperCAmelCase	:							List[str]
    UpperCAmelCase	:							Optional[List[int]]






class 						_A  (  _a  ):







    """simple docstring"""




    UpperCAmelCase	:							str






class 						_A  (  _a  ):







    """simple docstring"""




    UpperCAmelCase	:							Any






class 						_A  (  _a  ):







    """simple docstring"""




    @staticmethod
    def      __snake_case							( __UpperCAmelCase    :   ArgumentParser):
      a						:		str											=      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=8888    ,			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    :   Optional[Any]    ,			__UpperCAmelCase    :   Pipeline    ,			__UpperCAmelCase    :   str    ,			__UpperCAmelCase    :   int    ,			__UpperCAmelCase    :   int):
      a						:		Optional[int]											=      pipeline

      a						:		List[str]											=      host
      a						:		Union[str, Any]											=      port
      a						:		Dict											=      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						:		Any											=      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      __snake_case							( self    :   Dict):
      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    :   Dict    ,			__UpperCAmelCase    :   str = Body(__UpperCAmelCase    ,			embed=__UpperCAmelCase)    ,			__UpperCAmelCase    :   bool = Body(__UpperCAmelCase    ,			embed=__UpperCAmelCase)):
      try:
        a						:		Optional[Any]											=      self._pipeline.tokenizer.tokenize(__UpperCAmelCase)

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

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

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










"""simple docstring"""







import datasets


__lowercase    =				"""\
@InProceedings{conneau2018xnli,
  author = \"Conneau, Alexis
                 and Rinott, Ruty
                 and Lample, Guillaume
                 and Williams, Adina
                 and Bowman, Samuel R.
                 and Schwenk, Holger
                 and Stoyanov, Veselin\",
  title = \"XNLI: Evaluating Cross-lingual Sentence Representations\",
  booktitle = \"Proceedings of the 2018 Conference on Empirical Methods
               in Natural Language Processing\",
  year = \"2018\",
  publisher = \"Association for Computational Linguistics\",
  location = \"Brussels, Belgium\",
}
"""

__lowercase    =				"""\
XNLI is a subset of a few thousand examples from MNLI which has been translated
into a 14 different languages (some low-ish resource). As with MNLI, the goal is
to predict textual entailment (does sentence A imply/contradict/neither sentence
B) and is a classification task (given two sentences, predict one of three
labels).
"""

__lowercase    =				"""
Computes XNLI score which is just simple accuracy.
Args:
    predictions: Predicted labels.
    references: Ground truth labels.
Returns:
    'accuracy': accuracy
Examples:

    >>> predictions = [0, 1]
    >>> references = [0, 1]
    >>> xnli_metric = datasets.load_metric(\"xnli\")
    >>> results = xnli_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'accuracy': 1.0}
"""


def 					lowercase      (    A_  ,       A_	)->				List[str]:
  '''simple docstring'''







  return (preds == labels).mean()







@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION   ,_KWARGS_DESCRIPTION  )
class 						_A  (  datasets.Metric  ):







    """simple docstring"""




    def      __snake_case							( self    :   List[str]):
      return datasets.MetricInfo(
          description=_DESCRIPTION    ,			citation=_CITATION    ,			inputs_description=_KWARGS_DESCRIPTION    ,			features=datasets.Features(
              {
                  "predictions": datasets.Value("int64" if self.config_name != "sts-b" else "float32"),
                  "references": datasets.Value("int64" if self.config_name != "sts-b" else "float32"),
              })    ,			codebase_urls=[]    ,			reference_urls=[]    ,			format="numpy"    ,			)






    def      __snake_case							( self    :   str    ,			__UpperCAmelCase    :   List[Any]    ,			__UpperCAmelCase    :   Tuple):
      return {"accuracy": simple_accuracy(__UpperCAmelCase    ,			__UpperCAmelCase)}






