


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							: str      =   True
except (ImportError, AttributeError):
						snake_case							: Dict      =   object






						def 							A     (				*__lowerCAmelCase							:       Any      ,					**__lowerCAmelCase							:       Optional[Any]     ):
													pass






						snake_case							: Dict      =   False


snake_case							: Dict      =   logging.get_logger('''transformers-cli/serving''')






def 							A     (				__lowerCAmelCase							:       Namespace     ):
							a__    =  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  snake_case_     (lowerCamelCase_   ):
					UpperCAmelCase__      :					dict







class  snake_case_     (lowerCamelCase_   ):
					UpperCAmelCase__      :					List[str]
					UpperCAmelCase__      :					Optional[List[int]]







class  snake_case_     (lowerCamelCase_   ):
					UpperCAmelCase__      :					str







class  snake_case_     (lowerCamelCase_   ):
					UpperCAmelCase__      :					Any







class  snake_case_     (lowerCamelCase_   ):
					@staticmethod
					def 				lowerCamelCase__(   __snake_case     :ArgumentParser   )      ->					Any:
												a__    =  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=88_88    ,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     :int    ,__snake_case     :Pipeline    ,__snake_case     :str    ,__snake_case     :int    ,__snake_case     :int   )      ->					Optional[int]:
												a__    =  pipeline

												a__    =  host
												a__    =  port
												a__    =  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__    =  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=6_00    ,)
					def 				lowerCamelCase__(   self     :Union[str, Any]   )      ->					Optional[int]:
												run(self._app    ,host=self.host    ,port=self.port    ,workers=self.workers   )
					def 				lowerCamelCase__(   self     :List[Any]   )      ->					List[Any]:
												return ServeModelInfoResult(infos=vars(self._pipeline.model.config   )   )
					def 				lowerCamelCase__(   self     :str    ,__snake_case     :str = Body(__snake_case    ,embed=__snake_case   )    ,__snake_case     :bool = Body(__snake_case    ,embed=__snake_case   )   )      ->					Dict:
												try:
																			a__    =  self._pipeline.tokenizer.tokenize(__snake_case   )

																			if return_ids:
																										a__    =  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=5_00    ,detail={'model': '', 'error': str(__snake_case   )}   )
					def 				lowerCamelCase__(   self     :Union[str, Any]    ,__snake_case     :List[int] = Body(__snake_case    ,embed=__snake_case   )    ,__snake_case     :bool = Body(__snake_case    ,embed=__snake_case   )    ,__snake_case     :bool = Body(__snake_case    ,embed=__snake_case   )    ,)      ->					str:
												try:
																			a__    =  self._pipeline.tokenizer.decode(__snake_case    ,__snake_case    ,__snake_case   )
																			return ServeDeTokenizeResult(model=''    ,text=__snake_case   )
												except Exception as e:
																			raise HTTPException(status_code=5_00    ,detail={'model': '', 'error': str(__snake_case   )}   )



					async def 				lowerCamelCase__(   self     :Tuple    ,__snake_case     :List[Any]=Body(__snake_case    ,embed=__snake_case   )   )      ->					Dict:

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

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








# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re

from ..utils import cached_file


# docstyle-ignore
snake_case							: Dict      =   '''
Human: <<task>>

Assistant: '''


snake_case							: Optional[int]      =   '''huggingface-tools/default-prompts'''
snake_case							: Tuple      =   {'''chat''': '''chat_prompt_template.txt''', '''run''': '''run_prompt_template.txt'''}






def 							__lowercase     (				__lowerCAmelCase							:       Any      ,					__lowerCAmelCase							:       Union[str, Any]      ,					__lowerCAmelCase							:       Any="run"     ):
							if prompt_or_repo_id is None:
														a__    =  DEFAULT_PROMPTS_REPO

							# prompt is considered a repo ID when it does not contain any kind of space
							if re.search('\\s'      ,					__lowerCAmelCase     ) is not None:
														return prompt_or_repo_id

							a__    =  cached_file(
							    __lowerCAmelCase      ,					PROMPT_FILES[mode]      ,					repo_type='dataset'      ,					user_agent={'agent': agent_name}     )
							with open(__lowerCAmelCase      ,					'r'      ,					encoding='utf-8'     ) as f:
														return f.read()




