

'''simple docstring'''


import os
import sys
import warnings
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen
from ..table import array_cast
from ..utils.file_utils import is_local_path
from ..utils.py_utils import first_non_null_value, no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
		import PIL.Image

		from .features import FeatureType


lowerCAmelCase__						:       Optional[List[str]]										=						None
lowerCAmelCase__						:       Dict										=						"""<""" if sys.byteorder == """little""" else """>"""
# Origin: https://github.com/python-pillow/Pillow/blob/698951e19e19972aeed56df686868f1329981c12/src/PIL/Image.py#L3126 minus "|i1" which values are not preserved correctly when saving and loading an image
lowerCAmelCase__						:       Optional[Any]										=						[
    np.dtype("""|b1"""),
    np.dtype("""|u1"""),
    np.dtype("""<u2"""),
    np.dtype(""">u2"""),
    np.dtype("""<i2"""),
    np.dtype(""">i2"""),
    np.dtype("""<u4"""),
    np.dtype(""">u4"""),
    np.dtype("""<i4"""),
    np.dtype(""">i4"""),
    np.dtype("""<f4"""),
    np.dtype(""">f4"""),
    np.dtype("""<f8"""),
    np.dtype(""">f8"""),
]




@dataclass
class     a			:






						"""simple docstring"""




						__UpperCAmelCase			     =    True
						__UpperCAmelCase			     =    None
						# Automatically constructed
						__UpperCAmelCase			     =    """PIL.Image.Image"""
						__UpperCAmelCase			     =    pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}       )
						__UpperCAmelCase			     =    field(default="""Image"""			,			init=SCREAMING_SNAKE_CASE__			,			repr=SCREAMING_SNAKE_CASE__       )







						def __call__(							self					:      List[str]  ):



								'''simple docstring'''



								return self.pa_type







						def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      Tuple  ):



								'''simple docstring'''



								if config.PIL_AVAILABLE:
										import PIL.Image
								else:
										raise ImportError('''To support encoding images, please install \'Pillow\'.'''  )

								if isinstance(snake_case__			,			snake_case__  ):
										snake_case__						:       Optional[Any]       						=  np.array(snake_case__  )

								if isinstance(snake_case__			,			snake_case__  ):
										return {"path": value, "bytes": None}
								elif isinstance(snake_case__			,			snake_case__  ):
										return {"path": None, "bytes": value}
								elif isinstance(snake_case__			,			np.ndarray  ):
										# convert the image array to PNG/TIFF bytes
										return encode_np_array(snake_case__  )
								elif isinstance(snake_case__			,			PIL.Image.Image  ):
										# convert the PIL image to bytes (default format is PNG/TIFF)
										return encode_pil_image(snake_case__  )
								elif value.get('''path'''  ) is not None and os.path.isfile(value['''path''']  ):
										# we set "bytes": None to not duplicate the data if they're already available locally
										return {"bytes": None, "path": value.get('''path'''  )}
								elif value.get('''bytes'''  ) is not None or value.get('''path'''  ) is not None:
										# store the image bytes, and path is used to infer the image format using the file extension
										return {"bytes": value.get('''bytes'''  ), "path": value.get('''path'''  )}
								else:
										raise ValueError(
										    F"""An image sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}."""  )







						def       __magic_name__							(							self					:      Union[str, Any]			,			snake_case_					:      List[str]			,			snake_case_					:      Optional[int]=None  ):



								'''simple docstring'''



								if not self.decode:
										raise RuntimeError('''Decoding is disabled for this feature. Please use Image(decode=True) instead.'''  )

								if config.PIL_AVAILABLE:
										import PIL.Image
								else:
										raise ImportError('''To support decoding images, please install \'Pillow\'.'''  )

								if token_per_repo_id is None:
										snake_case__						:       List[Any]       						=  {}

								snake_case__						:       str       						=  value["path"], value["bytes"]
								if bytes_ is None:
										if path is None:
												raise ValueError(F"""An image should have one of \'path\' or \'bytes\' but both are None in {value}."""  )
										else:
												if is_local_path(snake_case__  ):
														snake_case__						:       int       						=  PIL.Image.open(snake_case__  )
												else:
														snake_case__						:       List[str]       						=  path.split('''::'''  )[-1]
														try:
																snake_case__						:       List[Any]       						=  string_to_dict(snake_case__			,			config.HUB_DATASETS_URL  )["repo_id"]
																snake_case__						:       Tuple       						=  token_per_repo_id.get(snake_case__  )
														except ValueError:
																snake_case__						:       List[Any]       						=  None
														with xopen(snake_case__			,			'''rb'''			,			use_auth_token=snake_case__  ) as f:
																snake_case__						:       Optional[Any]       						=  BytesIO(f.read()  )
														snake_case__						:       Dict       						=  PIL.Image.open(bytes_  )
								else:
										snake_case__						:       Any       						=  PIL.Image.open(BytesIO(bytes_  )  )
								image.load()  # to avoid "Too many open files" errors
								return image







						def       __magic_name__							(							self					:      Any  ):



								'''simple docstring'''



								from .features import Value

								return (
								    self
								    if self.decode
								    else {
								        "bytes": Value('''binary'''  ),
								        "path": Value('''string'''  ),
								    }
								)







						def       __magic_name__							(							self					:      Union[str, Any]			,			snake_case_					:      str  ):



								'''simple docstring'''



								if pa.types.is_string(storage.type  ):
										snake_case__						:       Optional[int]       						=  pa.array([None] * len(snake_case__  )			,			type=pa.binary()  )
										snake_case__						:       Optional[int]       						=  pa.StructArray.from_arrays([bytes_array, storage]			,			['''bytes''', '''path''']			,			mask=storage.is_null()  )
								elif pa.types.is_binary(storage.type  ):
										snake_case__						:       Optional[Any]       						=  pa.array([None] * len(snake_case__  )			,			type=pa.string()  )
										snake_case__						:       List[str]       						=  pa.StructArray.from_arrays([storage, path_array]			,			['''bytes''', '''path''']			,			mask=storage.is_null()  )
								elif pa.types.is_struct(storage.type  ):
										if storage.type.get_field_index('''bytes'''  ) >= 0:
												snake_case__						:       Union[str, Any]       						=  storage.field('''bytes'''  )
										else:
												snake_case__						:       Union[str, Any]       						=  pa.array([None] * len(snake_case__  )			,			type=pa.binary()  )
										if storage.type.get_field_index('''path'''  ) >= 0:
												snake_case__						:       str       						=  storage.field('''path'''  )
										else:
												snake_case__						:       int       						=  pa.array([None] * len(snake_case__  )			,			type=pa.string()  )
										snake_case__						:       List[Any]       						=  pa.StructArray.from_arrays([bytes_array, path_array]			,			['''bytes''', '''path''']			,			mask=storage.is_null()  )
								elif pa.types.is_list(storage.type  ):
										snake_case__						:       Optional[int]       						=  pa.array(
										    [encode_np_array(np.array(snake_case__  )  )['''bytes'''] if arr is not None else None for arr in storage.to_pylist()]			,			type=pa.binary()			,			)
										snake_case__						:       List[Any]       						=  pa.array([None] * len(snake_case__  )			,			type=pa.string()  )
										snake_case__						:       List[Any]       						=  pa.StructArray.from_arrays(
										    [bytes_array, path_array]			,			['''bytes''', '''path''']			,			mask=bytes_array.is_null()  )
								return array_cast(snake_case__			,			self.pa_type  )







						def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      Dict  ):



								'''simple docstring'''



								@no_op_if_value_is_null
								def path_to_bytes(snake_case_					:      Tuple  ):
										with xopen(snake_case__			,			'''rb'''  ) as f:
												snake_case__						:       Optional[int]       						=  f.read()
										return bytes_

								snake_case__						:       List[str]       						=  pa.array(
								    [
								        (path_to_bytes(x['''path''']  ) if x['''bytes'''] is None else x['''bytes''']) if x is not None else None
								        for x in storage.to_pylist()
								    ]			,			type=pa.binary()			,			)
								snake_case__						:       Optional[int]       						=  pa.array(
								    [os.path.basename(snake_case__  ) if path is not None else None for path in storage.field('''path'''  ).to_pylist()]			,			type=pa.string()			,			)
								snake_case__						:       str       						=  pa.StructArray.from_arrays([bytes_array, path_array]			,			['''bytes''', '''path''']			,			mask=bytes_array.is_null()  )
								return array_cast(snake_case__			,			self.pa_type  )





def 	_a       (				):

		"""simple docstring"""

		if config.PIL_AVAILABLE:
				import PIL.Image
		else:
				raise ImportError('''To support encoding images, please install \'Pillow\'.'''     )

		global _IMAGE_COMPRESSION_FORMATS
		if _IMAGE_COMPRESSION_FORMATS is None:
				PIL.Image.init()
				snake_case__						:       Any       						=  list(set(PIL.Image.OPEN.keys()     ) & set(PIL.Image.SAVE.keys()     )     )
		return _IMAGE_COMPRESSION_FORMATS




def 	_a       (				__lowerCAmelCase		:      Any     ):

		"""simple docstring"""

		snake_case__						:       Union[str, Any]       						=  BytesIO()
		if image.format in list_image_compression_formats():
				snake_case__						:       Dict       						=  image.format
		else:
				snake_case__						:       List[str]       						=  "PNG" if image.mode in ["1", "L", "LA", "RGB", "RGBA"] else "TIFF"
		image.save(__lowerCAmelCase   ,							format=__lowerCAmelCase     )
		return buffer.getvalue()




def 	_a       (				__lowerCAmelCase		:      Union[str, Any]     ):

		"""simple docstring"""

		if hasattr(__lowerCAmelCase   ,							'''filename'''     ) and image.filename != "":
				return {"path": image.filename, "bytes": None}
		else:
				return {"path": None, "bytes": image_to_bytes(__lowerCAmelCase     )}




def 	_a       (				__lowerCAmelCase		:      List[str]     ):

		"""simple docstring"""

		if config.PIL_AVAILABLE:
				import PIL.Image
		else:
				raise ImportError('''To support encoding images, please install \'Pillow\'.'''     )

		snake_case__						:       List[str]       						=  array.dtype
		snake_case__						:       Optional[int]       						=  dtype.byteorder if dtype.byteorder != "=" else _NATIVE_BYTEORDER
		snake_case__						:       Optional[int]       						=  dtype.kind
		snake_case__						:       List[str]       						=  dtype.itemsize

		snake_case__						:       str       						=  None

		# Multi-channel array case (only np.dtype("|u1") is allowed)
		if array.shape[2:]:
				snake_case__						:       Optional[Any]       						=  np.dtype('''|u1'''     )
				if dtype_kind not in ["u", "i"]:
						raise TypeError(
						    F"""Unsupported array dtype {dtype} for image encoding. Only {dest_dtype} is supported for multi-channel arrays."""     )
				if dtype is not dest_dtype:
						warnings.warn(F"""Downcasting array dtype {dtype} to {dest_dtype} to be compatible with \'Pillow\'"""     )
    # Exact match
		elif dtype in _VALID_IMAGE_ARRAY_DTPYES:
				snake_case__						:       Any       						=  dtype
		else:  # Downcast the type within the kind (np.can_cast(from_type, to_type, casting="same_kind") doesn't behave as expected, so do it manually)
				while dtype_itemsize >= 1:
						snake_case__						:       List[Any]       						=  dtype_byteorder + dtype_kind + str(__lowerCAmelCase     )
						snake_case__						:       Dict       						=  np.dtype(__lowerCAmelCase     )
						if dest_dtype in _VALID_IMAGE_ARRAY_DTPYES:
								warnings.warn(F"""Downcasting array dtype {dtype} to {dest_dtype} to be compatible with \'Pillow\'"""     )
								break
						else:
								dtype_itemsize //= 2

		if dest_dtype is None:
				raise TypeError(
				    F"""Cannot convert dtype {dtype} to a valid image dtype. Valid image dtypes: {_VALID_IMAGE_ARRAY_DTPYES}"""     )
		snake_case__						:       Tuple       						=  PIL.Image.fromarray(array.astype(__lowerCAmelCase     )     )
		return {"path": None, "bytes": image_to_bytes(__lowerCAmelCase     )}





def 	_a       (				__lowerCAmelCase		:      Optional[Any]     ):

		"""simple docstring"""

		if config.PIL_AVAILABLE:
				import PIL.Image
		else:
				raise ImportError('''To support encoding images, please install \'Pillow\'.'''     )

		if objs:
				snake_case__						:       Any       						=  first_non_null_value(__lowerCAmelCase     )
				if isinstance(__lowerCAmelCase   ,							__lowerCAmelCase     ):
						return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
				if isinstance(__lowerCAmelCase   ,							np.ndarray     ):
						snake_case__						:       Optional[Any]       						=  no_op_if_value_is_null(__lowerCAmelCase     )
						return [obj_to_image_dict_func(__lowerCAmelCase     ) for obj in objs]
				elif isinstance(__lowerCAmelCase   ,							PIL.Image.Image     ):
						snake_case__						:       List[str]       						=  no_op_if_value_is_null(__lowerCAmelCase     )
						return [obj_to_image_dict_func(__lowerCAmelCase     ) for obj in objs]
				else:
						return objs
		else:
				return objs






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

					lowerCAmelCase__						:       Any										=						True
except (ImportError, AttributeError):
					lowerCAmelCase__						:       Dict										=						object




					def 	_a       (				*__lowerCAmelCase		:      List[str]   ,							**__lowerCAmelCase		:      Tuple     ):

							"""simple docstring"""

							pass





					lowerCAmelCase__						:       str										=						False


lowerCAmelCase__						:       List[str]										=						logging.get_logger("""transformers-cli/serving""")




def 	_a       (				__lowerCAmelCase		:      Namespace     ):

		"""simple docstring"""

		snake_case__						:       Union[str, 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     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42
						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						__UpperCAmelCase			     =    42





class     a			(				SCREAMING_SNAKE_CASE       ):






						"""simple docstring"""




						@staticmethod
						def       __magic_name__							(							snake_case_					:      ArgumentParser  ):



								'''simple docstring'''



								snake_case__						:       Optional[Any]       						=  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=8_8_8_8			,			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					:      Union[str, Any]			,			snake_case_					:      Pipeline			,			snake_case_					:      str			,			snake_case_					:      int			,			snake_case_					:      int  ):



								'''simple docstring'''



								snake_case__						:       Any       						=  pipeline

								snake_case__						:       Tuple       						=  host
								snake_case__						:       Optional[Any]       						=  port
								snake_case__						:       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}"""  )
										snake_case__						:       str       						=  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_0_0			,			)







						def       __magic_name__							(							self					:      str  ):



								'''simple docstring'''



								run(self._app			,			host=self.host			,			port=self.port			,			workers=self.workers  )







						def       __magic_name__							(							self					:      Optional[Any]  ):



								'''simple docstring'''



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







						def       __magic_name__							(							self					:      List[str]			,			snake_case_					:      str = Body(snake_case_			,			embed=snake_case_  )			,			snake_case_					:      bool = Body(snake_case_			,			embed=snake_case_  )  ):



								'''simple docstring'''



								try:
										snake_case__						:       Optional[Any]       						=  self._pipeline.tokenizer.tokenize(snake_case_  )

										if return_ids:
												snake_case__						:       Optional[int]       						=  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_0_0			,			detail={'''model''': '''''', '''error''': str(snake_case_  )}  )







						def       __magic_name__							(							self					:      List[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_  )			,			):



								'''simple docstring'''



								try:
										snake_case__						:       Optional[int]       						=  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_0_0			,			detail={'''model''': '''''', '''error''': str(snake_case_  )}  )







						async def       __magic_name__							(							self					:      Tuple			,			snake_case_					:      List[str]=Body(snake_case_			,			embed=snake_case_  )  ):



								'''simple docstring'''



								if len(snake_case_  ) == 0:
										return ServeForwardResult(output=[]			,			attention=[]  )

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



