


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

   SCREAMING_SNAKE_CASE		:List[Any]     						=   True
except (ImportError, AttributeError):
   SCREAMING_SNAKE_CASE		:Tuple     						=   object



   def 	_lowerCAmelCase  (							*lowerCAmelCase_							:Tuple						,  **lowerCAmelCase_							:Optional[Any]							)->Dict:
      '''simple docstring'''






      pass



   SCREAMING_SNAKE_CASE		:str     						=   False


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



def 	_lowerCAmelCase  (							lowerCAmelCase_							:Namespace							)->Optional[int]:
   '''simple docstring'''






   snake_case_    			=    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   (       _a    ):






  """simple docstring"""





  _SCREAMING_SNAKE_CASE									=				42






class 			__lowerCAmelCase   (       _a    ):






  """simple docstring"""





  _SCREAMING_SNAKE_CASE									=				42
  _SCREAMING_SNAKE_CASE									=				42






class 			__lowerCAmelCase   (       _a    ):






  """simple docstring"""





  _SCREAMING_SNAKE_CASE									=				42






class 			__lowerCAmelCase   (       _a    ):






  """simple docstring"""





  _SCREAMING_SNAKE_CASE									=				42






class 			__lowerCAmelCase   (       _a    ):






  """simple docstring"""





  @staticmethod
  def     lowerCAmelCase__     (   _lowerCAmelCase				:				ArgumentParser     )      ->       Tuple:




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



  def __init__(   self				:				Optional[int]			, _lowerCAmelCase				:				Pipeline			, _lowerCAmelCase				:				str			, _lowerCAmelCase				:				int			, _lowerCAmelCase				:				int     )      ->       List[Any]:




     """simple docstring"""
     snake_case_    			=    pipeline

     snake_case_    			=    host
     snake_case_    			=    port
     snake_case_    			=    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}'''     )
        snake_case_    			=    FastAPI(
            routes=[
                APIRoute(
                    "/"			, self.model_info			, response_model=__lowerCamelCase			, response_class=__lowerCamelCase			, methods=["GET"]			, ),
                APIRoute(
                    "/tokenize"			, self.tokenize			, response_model=__lowerCamelCase			, response_class=__lowerCamelCase			, methods=["POST"]			, ),
                APIRoute(
                    "/detokenize"			, self.detokenize			, response_model=__lowerCamelCase			, response_class=__lowerCamelCase			, methods=["POST"]			, ),
                APIRoute(
                    "/forward"			, self.forward			, response_model=__lowerCamelCase			, response_class=__lowerCamelCase			, methods=["POST"]			, ),
            ]			, timeout=6_0_0			, )



  def     lowerCAmelCase__     (   self				:				str     )      ->       Dict:




     """simple docstring"""
     run(self._app			, host=self.host			, port=self.port			, workers=self.workers     )



  def     lowerCAmelCase__     (   self				:				Optional[Any]     )      ->       List[Any]:




     """simple docstring"""
     return ServeModelInfoResult(infos=vars(self._pipeline.model.config     )     )



  def     lowerCAmelCase__     (   self				:				Optional[Any]			, _lowerCAmelCase				:				str = Body(__lowerCamelCase			, embed=__lowerCamelCase     )			, _lowerCAmelCase				:				bool = Body(__lowerCamelCase			, embed=__lowerCamelCase     )     )      ->       List[Any]:




     """simple docstring"""
     try:
        snake_case_    			=    self._pipeline.tokenizer.tokenize(__lowerCamelCase     )

        if return_ids:
           snake_case_    			=    self._pipeline.tokenizer.convert_tokens_to_ids(__lowerCamelCase     )
           return ServeTokenizeResult(tokens=__lowerCamelCase			, tokens_ids=__lowerCamelCase     )
        else:
           return ServeTokenizeResult(tokens=__lowerCamelCase     )

     except Exception as e:
        raise HTTPException(status_code=5_0_0			, detail={"model": "", "error": str(__lowerCamelCase     )}     )



  def     lowerCAmelCase__     (   self				:				Optional[int]			, _lowerCAmelCase				:				List[int] = Body(__lowerCamelCase			, embed=__lowerCamelCase     )			, _lowerCAmelCase				:				bool = Body(__lowerCamelCase			, embed=__lowerCamelCase     )			, _lowerCAmelCase				:				bool = Body(__lowerCamelCase			, embed=__lowerCamelCase     )			, )      ->       int:




     """simple docstring"""
     try:
        snake_case_    			=    self._pipeline.tokenizer.decode(__lowerCamelCase			, __lowerCamelCase			, __lowerCamelCase     )
        return ServeDeTokenizeResult(model=""			, text=__lowerCamelCase     )
     except Exception as e:
        raise HTTPException(status_code=5_0_0			, detail={"model": "", "error": str(__lowerCamelCase     )}     )



  async def     lowerCAmelCase__     (   self				:				Tuple			, _lowerCAmelCase				:				Dict=Body(__lowerCamelCase			, embed=__lowerCamelCase     )     )      ->       Optional[int]:




     """simple docstring"""

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

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












import copy
from dataclasses import dataclass, field
from typing import ClassVar, Dict

from ..features import Audio, Features, Value
from .base import TaskTemplate

@dataclass(frozen=_a      )
class     _SCREAMING_SNAKE_CASE       (			_a      ):
	snake_case__							:     str     	=  field(default="""automatic-speech-recognition"""		,   metadata={"""include_in_asdict_even_if_is_default""": True}      )
	snake_case__							:     ClassVar[Features]     	=  Features({"""audio""": Audio()}      )
	snake_case__							:     ClassVar[Features]     	=  Features({"""transcription""": Value("""string"""      )}      )
	snake_case__							:     str     	=  "audio"
	snake_case__							:     str     	=  "transcription"





	def 						_A (		self				:					List[str] ,  __lowerCamelCase				:					Dict	):
		if self.audio_column not in features:
			raise ValueError(F"""Column {self.audio_column} is not present in features."""	)
		if not isinstance(features[self.audio_column] ,  __lowerCamelCase	):
			raise ValueError(F"""Column {self.audio_column} is not an Audio type."""	)
		UpperCamelCase     :int          =	copy.deepcopy(self	)
		UpperCamelCase     :Any          =	self.input_schema.copy()
		UpperCamelCase     :List[str]          =	features[self.audio_column]
		UpperCamelCase     :List[Any]          =	input_schema
		return task_template





	@property
	def 						_A (		self				:					Optional[int]	):
		return {self.audio_column: "audio", self.transcription_column: "transcription"}
