


'''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_						:  Dict				  =						True
except (ImportError, AttributeError):
				UpperCAmelCase_						:  List[Any]				  =						object







				def      snake_case_  (    *SCREAMING_SNAKE_CASE__				, **SCREAMING_SNAKE_CASE__	):







								"""simple docstring"""





								pass







				UpperCAmelCase_						:  Optional[int]				  =						False


UpperCAmelCase_						:  str				  =						logging.get_logger('transformers-cli/serving')







def      snake_case_  (    SCREAMING_SNAKE_CASE__	):







				"""simple docstring"""





				_SCREAMING_SNAKE_CASE					:			Optional[int]							= 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        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							A_				: dict





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							A_				: List[str]
							A_				: Optional[List[int]]





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							A_				: str





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


							A_				: Any





class        lowercase__    (		_snake_case   ):





							'''simple docstring'''


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






							def __init__(       self       ,   __snake_case       ,   __snake_case       ,   __snake_case       ,   __snake_case    ):
											_SCREAMING_SNAKE_CASE					:			Optional[Any]							= pipeline

											_SCREAMING_SNAKE_CASE					:			Any							= host
											_SCREAMING_SNAKE_CASE					:			Dict							= port
											_SCREAMING_SNAKE_CASE					:			List[str]							= 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}"""    )
															_SCREAMING_SNAKE_CASE					:			Dict							= FastAPI(
															    routes=[
															        APIRoute(
															            """/"""       ,   self.model_info       ,   response_model=__snake_case       ,   response_class=__snake_case       ,   methods=["""GET"""]       ,   ),
															        APIRoute(
															            """/tokenize"""       ,   self.tokenize       ,   response_model=__snake_case       ,   response_class=__snake_case       ,   methods=["""POST"""]       ,   ),
															        APIRoute(
															            """/detokenize"""       ,   self.detokenize       ,   response_model=__snake_case       ,   response_class=__snake_case       ,   methods=["""POST"""]       ,   ),
															        APIRoute(
															            """/forward"""       ,   self.forward       ,   response_model=__snake_case       ,   response_class=__snake_case       ,   methods=["""POST"""]       ,   ),
															    ]       ,   timeout=600       ,   )






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






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






							def UpperCAmelCase_      (       self       ,   __snake_case = Body(__snake_case       ,   embed=__snake_case    )       ,   __snake_case = Body(__snake_case       ,   embed=__snake_case    )    ):
											try:
															_SCREAMING_SNAKE_CASE					:			List[str]							= self._pipeline.tokenizer.tokenize(__snake_case    )

															if return_ids:
																			_SCREAMING_SNAKE_CASE					:			str							= self._pipeline.tokenizer.convert_tokens_to_ids(__snake_case    )
																			return ServeTokenizeResult(tokens=__snake_case       ,   tokens_ids=__snake_case    )
															else:
																			return ServeTokenizeResult(tokens=__snake_case    )

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






							def UpperCAmelCase_      (       self       ,   __snake_case = Body(__snake_case       ,   embed=__snake_case    )       ,   __snake_case = Body(__snake_case       ,   embed=__snake_case    )       ,   __snake_case = Body(__snake_case       ,   embed=__snake_case    )       ,   ):
											try:
															_SCREAMING_SNAKE_CASE					:			Tuple							= self._pipeline.tokenizer.decode(__snake_case       ,   __snake_case       ,   __snake_case    )
															return ServeDeTokenizeResult(model=""""""       ,   text=__snake_case    )
											except Exception as e:
															raise HTTPException(status_code=500       ,   detail={"""model""": """""", """error""": str(__snake_case    )}    )






							async def UpperCAmelCase_      (       self       ,   __snake_case=Body(__snake_case       ,   embed=__snake_case    )    ):

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

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








'''simple docstring'''


import os


# All paths are set with the intent you should run this script from the root of the repo with the command
# python utils/check_doctest_list.py
UpperCAmelCase_						:  List[Any]				  =						'.'


if __name__ == "__main__":
				UpperCAmelCase_						:  Any				  =						os.path.join(REPO_PATH, 'utils/documentation_tests.txt')
				UpperCAmelCase_						:  Dict				  =						[]
				UpperCAmelCase_						:  Union[str, Any]				  =						[]
				with open(doctest_file_path) as fp:
								for line in fp:
												UpperCAmelCase_						:  int				  =						line.strip()
												UpperCAmelCase_						:  str				  =						os.path.join(REPO_PATH, line)
												if not (os.path.isfile(path) or os.path.isdir(path)):
																non_existent_paths.append(line)
												all_paths.append(path)
				if len(non_existent_paths) > 0:
								UpperCAmelCase_						:  int				  =						'\n'.join(non_existent_paths)
								raise ValueError(F"`utils/documentation_tests.txt` contains non-existent paths:\n{non_existent_paths}")
				if all_paths != sorted(all_paths):
								raise ValueError('Files in `utils/documentation_tests.txt` are not in alphabetical order.')




