





import unittest

from diffusers.models.unet_ad_blocks import *  # noqa F403
from diffusers.utils import torch_device

from .test_unet_blocks_common import UNetBlockTesterMixin
class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Any       =	DownBlockaD  # noqa F405
		__a       :       Optional[Any]       =	"""down"""






		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									lowercase_	:   Optional[int]  				=     [-0.0232, -0.9869, 0.8054, -0.0637, -0.1688, -1.4264, 0.4470, -1.3394, 0.0904]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       int       =	ResnetDownsampleBlockaD  # noqa F405
		__a       :       List[Any]       =	"""down"""






		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									lowercase_	:   Optional[Any]  				=     [0.0710, 0.2410, -0.7320, -1.0757, -1.1343, 0.3540, -0.0133, -0.2576, 0.0948]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       List[Any]       =	AttnDownBlockaD  # noqa F405
		__a       :       List[str]       =	"""down"""






		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									lowercase_	:   Tuple  				=     [0.0636, 0.8964, -0.6234, -1.0131, 0.0844, 0.4935, 0.3437, 0.0911, -0.2957]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Optional[int]       =	CrossAttnDownBlockaD  # noqa F405
		__a       :       Tuple       =	"""down"""






		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									lowercase_				,		lowercase_	:   Tuple  				=     super().prepare_init_args_and_inputs_for_common()
									lowercase_	:   Any  				=     32
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									lowercase_	:   Optional[int]  				=     [0.2238, -0.7396, -0.2255, -0.3829, 0.1925, 1.1665, 0.0603, -0.7295, 0.1983]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       List[str]       =	SimpleCrossAttnDownBlockaD  # noqa F405
		__a       :       int       =	"""down"""






		@property
		def 						snake_case__				(							self  )       -> str:






									"""simple docstring"""
									return super().get_dummy_input(include_encoder_hidden_states=snake_case__  )






		def 						snake_case__				(							self  )       -> int:






									"""simple docstring"""
									lowercase_				,		lowercase_	:   Optional[Any]  				=     super().prepare_init_args_and_inputs_for_common()
									lowercase_	:   Union[str, Any]  				=     32
									return init_dict, inputs_dict






		@unittest.skipIf(torch_device == """mps""",   """MPS result is not consistent"""  )
		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									lowercase_	:   Optional[Any]  				=     [0.7921, -0.0992, -0.1962, -0.7695, -0.4242, 0.7804, 0.4737, 0.2765, 0.3338]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Dict       =	SkipDownBlockaD  # noqa F405
		__a       :       int       =	"""down"""






		@property
		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									return super().get_dummy_input(include_skip_sample=snake_case__  )






		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									lowercase_	:   List[str]  				=     [-0.0845, -0.2087, -0.2465, 0.0971, 0.1900, -0.0484, 0.2664, 0.4179, 0.5069]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Any       =	AttnSkipDownBlockaD  # noqa F405
		__a       :       List[Any]       =	"""down"""






		@property
		def 						snake_case__				(							self  )       -> str:






									"""simple docstring"""
									return super().get_dummy_input(include_skip_sample=snake_case__  )






		def 						snake_case__				(							self  )       -> Union[str, Any]:






									"""simple docstring"""
									lowercase_	:   List[Any]  				=     [0.5539, 0.1609, 0.4924, 0.0537, -0.1995, 0.4050, 0.0979, -0.2721, -0.0642]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Dict       =	DownEncoderBlockaD  # noqa F405
		__a       :       int       =	"""down"""






		@property
		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									return super().get_dummy_input(include_temb=snake_case__  )






		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									lowercase_	:   Optional[int]  				=     {
									    """in_channels""": 32,
									    """out_channels""": 32,
									}
									lowercase_	:   Dict  				=     self.dummy_input
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> Union[str, Any]:






									"""simple docstring"""
									lowercase_	:   List[str]  				=     [1.1102, 0.5302, 0.4872, -0.0023, -0.8042, 0.0483, -0.3489, -0.5632, 0.7626]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       str       =	AttnDownEncoderBlockaD  # noqa F405
		__a       :       int       =	"""down"""






		@property
		def 						snake_case__				(							self  )       -> str:






									"""simple docstring"""
									return super().get_dummy_input(include_temb=snake_case__  )






		def 						snake_case__				(							self  )       -> Any:






									"""simple docstring"""
									lowercase_	:   Optional[Any]  				=     {
									    """in_channels""": 32,
									    """out_channels""": 32,
									}
									lowercase_	:   int  				=     self.dummy_input
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> List[Any]:






									"""simple docstring"""
									lowercase_	:   Any  				=     [0.8966, -0.1486, 0.8568, 0.8141, -0.9046, -0.1342, -0.0972, -0.7417, 0.1538]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Union[str, Any]       =	UNetMidBlockaD  # noqa F405
		__a       :       List[Any]       =	"""mid"""






		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									lowercase_	:   Tuple  				=     {
									    """in_channels""": 32,
									    """temb_channels""": 1_28,
									}
									lowercase_	:   Dict  				=     self.dummy_input
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> Tuple:






									"""simple docstring"""
									lowercase_	:   List[Any]  				=     [-0.1062, 1.7248, 0.3494, 1.4569, -0.0910, -1.2421, -0.9984, 0.6736, 1.0028]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Optional[Any]       =	UNetMidBlockaDCrossAttn  # noqa F405
		__a       :       int       =	"""mid"""






		def 						snake_case__				(							self  )       -> Union[str, Any]:






									"""simple docstring"""
									lowercase_				,		lowercase_	:   List[str]  				=     super().prepare_init_args_and_inputs_for_common()
									lowercase_	:   Optional[Any]  				=     32
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> str:






									"""simple docstring"""
									lowercase_	:   Tuple  				=     [0.0187, 2.4220, 0.4484, 1.1203, -0.6121, -1.5122, -0.8270, 0.7851, 1.8335]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       List[str]       =	UNetMidBlockaDSimpleCrossAttn  # noqa F405
		__a       :       Optional[Any]       =	"""mid"""






		@property
		def 						snake_case__				(							self  )       -> Optional[int]:






									"""simple docstring"""
									return super().get_dummy_input(include_encoder_hidden_states=snake_case__  )






		def 						snake_case__				(							self  )       -> Union[str, Any]:






									"""simple docstring"""
									lowercase_				,		lowercase_	:   List[str]  				=     super().prepare_init_args_and_inputs_for_common()
									lowercase_	:   Dict  				=     32
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> Optional[int]:






									"""simple docstring"""
									lowercase_	:   List[str]  				=     [0.7143, 1.9974, 0.5448, 1.3977, 0.1282, -1.1237, -1.4238, 0.5530, 0.8880]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Optional[Any]       =	UpBlockaD  # noqa F405
		__a       :       Dict       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									return super().get_dummy_input(include_res_hidden_states_tuple=snake_case__  )






		def 						snake_case__				(							self  )       -> str:






									"""simple docstring"""
									lowercase_	:   Optional[Any]  				=     [-0.2041, -0.4165, -0.3022, 0.0041, -0.6628, -0.7053, 0.1928, -0.0325, 0.0523]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       str       =	ResnetUpsampleBlockaD  # noqa F405
		__a       :       Tuple       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> List[Any]:






									"""simple docstring"""
									return super().get_dummy_input(include_res_hidden_states_tuple=snake_case__  )






		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									lowercase_	:   str  				=     [0.2287, 0.3549, -0.1346, 0.4797, -0.1715, -0.9649, 0.7305, -0.5864, -0.6244]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Optional[int]       =	CrossAttnUpBlockaD  # noqa F405
		__a       :       List[Any]       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> str:






									"""simple docstring"""
									return super().get_dummy_input(include_res_hidden_states_tuple=snake_case__  )






		def 						snake_case__				(							self  )       -> List[Any]:






									"""simple docstring"""
									lowercase_				,		lowercase_	:   List[str]  				=     super().prepare_init_args_and_inputs_for_common()
									lowercase_	:   Optional[Any]  				=     32
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> List[str]:






									"""simple docstring"""
									lowercase_	:   Tuple  				=     [-0.1403, -0.3515, -0.0420, -0.1425, 0.3167, 0.5094, -0.2181, 0.5931, 0.5582]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Any       =	SimpleCrossAttnUpBlockaD  # noqa F405
		__a       :       Optional[int]       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									return super().get_dummy_input(include_res_hidden_states_tuple=snake_case__,   include_encoder_hidden_states=snake_case__  )






		def 						snake_case__				(							self  )       -> List[Any]:






									"""simple docstring"""
									lowercase_				,		lowercase_	:   Optional[Any]  				=     super().prepare_init_args_and_inputs_for_common()
									lowercase_	:   Union[str, Any]  				=     32
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> Tuple:






									"""simple docstring"""
									lowercase_	:   Any  				=     [0.2645, 0.1480, 0.0909, 0.8044, -0.9758, -0.9083, 0.0994, -1.1453, -0.7402]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Dict       =	AttnUpBlockaD  # noqa F405
		__a       :       Optional[Any]       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									return super().get_dummy_input(include_res_hidden_states_tuple=snake_case__  )






		@unittest.skipIf(torch_device == """mps""",   """MPS result is not consistent"""  )
		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									lowercase_	:   List[Any]  				=     [0.0979, 0.1326, 0.0021, 0.0659, 0.2249, 0.0059, 0.1132, 0.5952, 0.1033]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       int       =	SkipUpBlockaD  # noqa F405
		__a       :       Any       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> Optional[int]:






									"""simple docstring"""
									return super().get_dummy_input(include_res_hidden_states_tuple=snake_case__  )






		def 						snake_case__				(							self  )       -> Union[str, Any]:






									"""simple docstring"""
									lowercase_	:   int  				=     [-0.0893, -0.1234, -0.1506, -0.0332, 0.0123, -0.0211, 0.0566, 0.0143, 0.0362]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       Any       =	AttnSkipUpBlockaD  # noqa F405
		__a       :       Optional[Any]       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									return super().get_dummy_input(include_res_hidden_states_tuple=snake_case__  )






		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									lowercase_	:   List[Any]  				=     [0.0361, 0.0617, 0.2787, -0.0350, 0.0342, 0.3421, -0.0843, 0.0913, 0.3015]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       List[str]       =	UpDecoderBlockaD  # noqa F405
		__a       :       List[str]       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> str:






									"""simple docstring"""
									return super().get_dummy_input(include_temb=snake_case__  )






		def 						snake_case__				(							self  )       -> List[Any]:






									"""simple docstring"""
									lowercase_	:   Union[str, Any]  				=     {"""in_channels""": 32, """out_channels""": 32}

									lowercase_	:   str  				=     self.dummy_input
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> Tuple:






									"""simple docstring"""
									lowercase_	:   str  				=     [0.4404, 0.1998, -0.9886, -0.3320, -0.3128, -0.7034, -0.6955, -0.2338, -0.3137]
									super().test_output(snake_case__  )




class     UpperCamelCase__					(     lowerCamelCase__				, unittest.TestCase				):






		'''simple docstring'''






		__a       :       List[str]       =	AttnUpDecoderBlockaD  # noqa F405
		__a       :       Dict       =	"""up"""






		@property
		def 						snake_case__				(							self  )       -> Optional[Any]:






									"""simple docstring"""
									return super().get_dummy_input(include_temb=snake_case__  )






		def 						snake_case__				(							self  )       -> Dict:






									"""simple docstring"""
									lowercase_	:   int  				=     {"""in_channels""": 32, """out_channels""": 32}

									lowercase_	:   List[Any]  				=     self.dummy_input
									return init_dict, inputs_dict






		def 						snake_case__				(							self  )       -> Union[str, Any]:






									"""simple docstring"""
									lowercase_	:   Optional[Any]  				=     [0.6738, 0.4491, 0.1055, 1.0710, 0.7316, 0.3339, 0.3352, 0.1023, 0.3568]
									super().test_output(snake_case__  )






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_							=				True
except (ImportError, AttributeError):
							UpperCAmelCase_							=				object





							def  __magic_name__  (				*lowercase				,       **lowercase				)				-> Optional[Any]:

														"""simple docstring"""



														pass




							UpperCAmelCase_							=				False


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





def  __magic_name__  (				lowercase				)				-> int:

							"""simple docstring"""



							lowercase_	:   Any  				=     pipeline(
							    task=args.task				,       model=args.model if args.model else None				,       config=args.config				,       tokenizer=args.tokenizer				,       device=args.device				,       )
							return ServeCommand(lowercase				,       args.host				,       args.port				,       args.workers				)
class     UpperCamelCase__					(     lowerCamelCase__				):






		'''simple docstring'''






		__a       :       dict




class     UpperCamelCase__					(     lowerCamelCase__				):






		'''simple docstring'''






		__a       :       List[str]
		__a       :       Optional[List[int]]




class     UpperCamelCase__					(     lowerCamelCase__				):






		'''simple docstring'''






		__a       :       str




class     UpperCamelCase__					(     lowerCamelCase__				):






		'''simple docstring'''






		__a       :       Any




class     UpperCamelCase__					(     lowerCamelCase__				):






		'''simple docstring'''






		@staticmethod
		def 						snake_case__				(							snake_case__  )       -> Optional[int]:






									"""simple docstring"""
									lowercase_	:   List[str]  				=     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,   snake_case__,   snake_case__,   snake_case__,   snake_case__  )       -> Optional[Any]:






									"""simple docstring"""
									lowercase_	:   str  				=     pipeline

									lowercase_	:   List[str]  				=     host
									lowercase_	:   int  				=     port
									lowercase_	:   Optional[Any]  				=     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}"""  )
																lowercase_	:   Union[str, Any]  				=     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 						snake_case__				(							self  )       -> Union[str, Any]:






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






		def 						snake_case__				(							self  )       -> Dict:






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






		def 						snake_case__				(							self,   snake_case__ = Body(snake_case__,   embed=snake_case__  ),   snake_case__ = Body(snake_case__,   embed=snake_case__  )  )       -> Optional[int]:






									"""simple docstring"""
									try:
																lowercase_	:   Tuple  				=     self._pipeline.tokenizer.tokenize(snake_case__  )

																if return_ids:
																							lowercase_	:   Union[str, Any]  				=     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 						snake_case__				(							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__  ),   )       -> Dict:






									"""simple docstring"""
									try:
																lowercase_	:   List[Any]  				=     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 						snake_case__				(							self,   snake_case__=Body(snake_case__,   embed=snake_case__  )  )       -> Tuple:






									"""simple docstring"""

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

									try:
																# Forward through the model
																lowercase_	:   Optional[int]  				=     self._pipeline(snake_case__  )
																return ServeForwardResult(output=snake_case__  )
									except Exception as e:
																raise HTTPException(5_00,   {"""error""": str(snake_case__  )}  )