






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 	lowercase__ (	*__snake_case       :				Optional[int]	,			**__snake_case       :				List[Any]    ):


							'''simple docstring'''


							pass






		__UpperCAmelCase      				=		False


__UpperCAmelCase      				=		logging.get_logger('transformers-cli/serving')




def 	lowercase__ (	__snake_case       :				Union[str, Any]    ):


					'''simple docstring'''


					UpperCAmelCase_       : Optional[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     lowerCamelCase (_UpperCamelCase     ):







		'''simple docstring'''



		_snake_case       :       dict





class     lowerCamelCase (_UpperCamelCase     ):







		'''simple docstring'''



		_snake_case       :       List[str]
		_snake_case       :       Optional[List[int]]





class     lowerCamelCase (_UpperCamelCase     ):







		'''simple docstring'''



		_snake_case       :       str





class     lowerCamelCase (_UpperCamelCase     ):







		'''simple docstring'''



		_snake_case       :       Any





class     lowerCamelCase (_UpperCamelCase     ):







		'''simple docstring'''



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



		def __init__(				self							,						_UpperCamelCase							,						_UpperCamelCase							,						_UpperCamelCase							,						_UpperCamelCase  )      ->				List[str]:
							UpperCAmelCase_       : Optional[int]				=		pipeline

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



		def    __UpperCAmelCase     (				self  )      ->				Optional[int]:
							run(self._app							,						host=self.host							,						port=self.port							,						workers=self.workers  )



		def    __UpperCAmelCase     (				self  )      ->				str:
							return ServeModelInfoResult(infos=vars(self._pipeline.model.config  )  )



		def    __UpperCAmelCase     (				self							,						_UpperCamelCase = Body(lowercase__							,						embed=lowercase__  )							,						_UpperCamelCase = Body(lowercase__							,						embed=lowercase__  )  )      ->				Optional[int]:
							try:
												UpperCAmelCase_       : int				=		self._pipeline.tokenizer.tokenize(lowercase__  )

												if return_ids:
																	UpperCAmelCase_       : Dict				=		self._pipeline.tokenizer.convert_tokens_to_ids(lowercase__  )
																	return ServeTokenizeResult(tokens=lowercase__							,						tokens_ids=lowercase__  )
												else:
																	return ServeTokenizeResult(tokens=lowercase__  )

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



		def    __UpperCAmelCase     (				self							,						_UpperCamelCase = Body(lowercase__							,						embed=lowercase__  )							,						_UpperCamelCase = Body(lowercase__							,						embed=lowercase__  )							,						_UpperCamelCase = Body(lowercase__							,						embed=lowercase__  )							,						)      ->				List[Any]:
							try:
												UpperCAmelCase_       : Tuple				=		self._pipeline.tokenizer.decode(lowercase__							,						lowercase__							,						lowercase__  )
												return ServeDeTokenizeResult(model=''							,						text=lowercase__  )
							except Exception as e:
												raise HTTPException(status_code=5_0_0							,						detail={'model': '', 'error': str(lowercase__  )}  )



		async def    __UpperCAmelCase     (				self							,						_UpperCamelCase=Body(lowercase__							,						embed=lowercase__  )  )      ->				Union[str, Any]:

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

							try:
												# Forward through the model
												UpperCAmelCase_       : List[str]				=		self._pipeline(lowercase__  )
												return ServeForwardResult(output=lowercase__  )
							except Exception as e:
												raise HTTPException(5_0_0							,						{'error': str(lowercase__  )}  )










import inspect
import unittest

from transformers import DPTConfig
from transformers.file_utils import is_torch_available, is_vision_available
from transformers.models.auto import get_values
from transformers.testing_utils import require_torch, require_vision, slow, torch_device

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, _config_zero_init, floats_tensor, ids_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
				import torch
				from torch import nn

				from transformers import MODEL_MAPPING, DPTForDepthEstimation, DPTForSemanticSegmentation, DPTModel
				from transformers.models.dpt.modeling_dpt import DPT_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
				from PIL import Image

				from transformers import DPTImageProcessor




class 							lowerCamelCase   :


	def __init__(    self				,     lowercase__				,     lowercase__=2				,     lowercase__=3_2				,     lowercase__=1_6				,     lowercase__=3				,     lowercase__=True				,     lowercase__=True				,     lowercase__=3_2				,     lowercase__=4				,     lowercase__=[0, 1, 2, 3]				,     lowercase__=4				,     lowercase__=3_7				,     lowercase__="gelu"				,     lowercase__=0.1				,     lowercase__=0.1				,     lowercase__=0.0_2				,     lowercase__=3				,     lowercase__=[1, 3_8_4, 2_4, 2_4]				,     lowercase__=True				,     lowercase__=None				,     ):
							__UpperCAmelCase			:							Any	    =	parent
							__UpperCAmelCase			:							Tuple	    =	batch_size
							__UpperCAmelCase			:							Optional[int]	    =	image_size
							__UpperCAmelCase			:							Tuple	    =	patch_size
							__UpperCAmelCase			:							int	    =	num_channels
							__UpperCAmelCase			:							Optional[Any]	    =	is_training
							__UpperCAmelCase			:							Union[str, Any]	    =	use_labels
							__UpperCAmelCase			:							List[Any]	    =	hidden_size
							__UpperCAmelCase			:							Optional[Any]	    =	num_hidden_layers
							__UpperCAmelCase			:							Any	    =	backbone_out_indices
							__UpperCAmelCase			:							Union[str, Any]	    =	num_attention_heads
							__UpperCAmelCase			:							Dict	    =	intermediate_size
							__UpperCAmelCase			:							Dict	    =	hidden_act
							__UpperCAmelCase			:							Dict	    =	hidden_dropout_prob
							__UpperCAmelCase			:							List[Any]	    =	attention_probs_dropout_prob
							__UpperCAmelCase			:							List[Any]	    =	initializer_range
							__UpperCAmelCase			:							Union[str, Any]	    =	num_labels
							__UpperCAmelCase			:							List[Any]	    =	backbone_featmap_shape
							__UpperCAmelCase			:							Any	    =	scope
							__UpperCAmelCase			:							Optional[int]	    =	is_hybrid
							# sequence length of DPT = num_patches + 1 (we add 1 for the [CLS] token)
							__UpperCAmelCase			:							Optional[Any]	    =	(image_size // patch_size) ** 2
							__UpperCAmelCase			:							Any	    =	num_patches + 1


	def  A(    self):
							__UpperCAmelCase			:							Tuple	    =	floats_tensor([self.batch_size, self.num_channels, self.image_size, self.image_size])

							__UpperCAmelCase			:							List[str]	    =	None
							if self.use_labels:
													__UpperCAmelCase			:							Dict	    =	ids_tensor([self.batch_size, self.image_size, self.image_size]				,     self.num_labels)

							__UpperCAmelCase			:							Dict	    =	self.get_config()

							return config, pixel_values, labels


	def  A(    self):
							__UpperCAmelCase			:							Dict	    =	{
							    '''global_padding''': '''same''',
							    '''layer_type''': '''bottleneck''',
							    '''depths''': [3, 4, 9],
							    '''out_features''': ['''stage1''', '''stage2''', '''stage3'''],
							    '''embedding_dynamic_padding''': True,
							    '''hidden_sizes''': [9_6, 1_9_2, 3_8_4, 7_6_8],
							    '''num_groups''': 2,
							}

							return DPTConfig(
							    image_size=self.image_size				,     patch_size=self.patch_size				,     num_channels=self.num_channels				,     hidden_size=self.hidden_size				,     num_hidden_layers=self.num_hidden_layers				,     backbone_out_indices=self.backbone_out_indices				,     num_attention_heads=self.num_attention_heads				,     intermediate_size=self.intermediate_size				,     hidden_act=self.hidden_act				,     hidden_dropout_prob=self.hidden_dropout_prob				,     attention_probs_dropout_prob=self.attention_probs_dropout_prob				,     is_decoder=lowercase__				,     initializer_range=self.initializer_range				,     is_hybrid=self.is_hybrid				,     backbone_config=lowercase__				,     backbone_featmap_shape=self.backbone_featmap_shape				,     )


	def  A(    self				,     lowercase__				,     lowercase__				,     lowercase__):
							__UpperCAmelCase			:							List[str]	    =	DPTModel(config=lowercase__)
							model.to(lowercase__)
							model.eval()
							__UpperCAmelCase			:							Dict	    =	model(lowercase__)
							self.parent.assertEqual(result.last_hidden_state.shape				,     (self.batch_size, self.seq_length, self.hidden_size))


	def  A(    self				,     lowercase__				,     lowercase__				,     lowercase__):
							__UpperCAmelCase			:							List[str]	    =	self.num_labels
							__UpperCAmelCase			:							Optional[Any]	    =	DPTForDepthEstimation(lowercase__)
							model.to(lowercase__)
							model.eval()
							__UpperCAmelCase			:							str	    =	model(lowercase__)
							self.parent.assertEqual(result.predicted_depth.shape				,     (self.batch_size, self.image_size, self.image_size))


	def  A(    self				,     lowercase__				,     lowercase__				,     lowercase__):
							__UpperCAmelCase			:							Tuple	    =	self.num_labels
							__UpperCAmelCase			:							Optional[int]	    =	DPTForSemanticSegmentation(lowercase__)
							model.to(lowercase__)
							model.eval()
							__UpperCAmelCase			:							str	    =	model(lowercase__				,     labels=lowercase__)
							self.parent.assertEqual(
							    result.logits.shape				,     (self.batch_size, self.num_labels, self.image_size, self.image_size))


	def  A(    self):
							__UpperCAmelCase			:							int	    =	self.prepare_config_and_inputs()
							__UpperCAmelCase    ,      __UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[int]	    =	config_and_inputs
							__UpperCAmelCase			:							Union[str, Any]	    =	{'''pixel_values''': pixel_values}
							return config, inputs_dict






@require_torch
class 							lowerCamelCase   (  _UpperCamelCase							,						_UpperCamelCase							,						unittest.TestCase ):
	_lowerCAmelCase     :		Union[str, Any]										=						(DPTModel, DPTForDepthEstimation, DPTForSemanticSegmentation) if is_torch_available() else ()
	_lowerCAmelCase     :		Optional[int]										=						(
	    {
	        '''depth-estimation''': DPTForDepthEstimation,
	        '''feature-extraction''': DPTModel,
	        '''image-segmentation''': DPTForSemanticSegmentation,
	    }
	    if is_torch_available()
	    else {}
	)

	_lowerCAmelCase     :		Any										=						False
	_lowerCAmelCase     :		str										=						False
	_lowerCAmelCase     :		List[Any]										=						False


	def  A(    self):
							__UpperCAmelCase			:							Any	    =	DPTModelTester(self)
							__UpperCAmelCase			:							str	    =	ConfigTester(self				,     config_class=lowercase__				,     has_text_modality=lowercase__				,     hidden_size=3_7)


	def  A(    self):
							self.config_tester.run_common_tests()


	@unittest.skip(reason='''DPT does not use inputs_embeds''')
	def  A(    self):
							pass


	def  A(    self):
							__UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[Any]	    =	self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
													__UpperCAmelCase			:							str	    =	model_class(lowercase__)
													self.assertIsInstance(model.get_input_embeddings()				,     (nn.Module))
													__UpperCAmelCase			:							Tuple	    =	model.get_output_embeddings()
													self.assertTrue(x is None or isinstance(lowercase__				,     nn.Linear))


	def  A(    self):
							__UpperCAmelCase    ,      __UpperCAmelCase			:							Any	    =	self.model_tester.prepare_config_and_inputs_for_common()

							for model_class in self.all_model_classes:
													__UpperCAmelCase			:							Optional[int]	    =	model_class(lowercase__)
													__UpperCAmelCase			:							int	    =	inspect.signature(model.forward)
													# signature.parameters is an OrderedDict => so arg_names order is deterministic
													__UpperCAmelCase			:							List[str]	    =	[*signature.parameters.keys()]

													__UpperCAmelCase			:							str	    =	['''pixel_values''']
													self.assertListEqual(arg_names[:1]				,     lowercase__)


	def  A(    self):
							__UpperCAmelCase			:							Tuple	    =	self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_model(*lowercase__)


	def  A(    self):
							__UpperCAmelCase			:							List[str]	    =	self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_depth_estimation(*lowercase__)


	def  A(    self):
							__UpperCAmelCase			:							Dict	    =	self.model_tester.prepare_config_and_inputs()
							self.model_tester.create_and_check_for_semantic_segmentation(*lowercase__)


	def  A(    self):
							for model_class in self.all_model_classes:
													if model_class.__name__ == "DPTForDepthEstimation":
																			continue

													__UpperCAmelCase    ,      __UpperCAmelCase			:							str	    =	self.model_tester.prepare_config_and_inputs_for_common()
													__UpperCAmelCase			:							Any	    =	True

													if model_class in get_values(lowercase__):
																			continue

													__UpperCAmelCase			:							List[Any]	    =	model_class(lowercase__)
													model.to(lowercase__)
													model.train()
													__UpperCAmelCase			:							Any	    =	self._prepare_for_class(lowercase__				,     lowercase__				,     return_labels=lowercase__)
													__UpperCAmelCase			:							Any	    =	model(**lowercase__).loss
													loss.backward()


	def  A(    self):
							for model_class in self.all_model_classes:
													if model_class.__name__ == "DPTForDepthEstimation":
																			continue

													__UpperCAmelCase    ,      __UpperCAmelCase			:							int	    =	self.model_tester.prepare_config_and_inputs_for_common()
													__UpperCAmelCase			:							Optional[int]	    =	False
													__UpperCAmelCase			:							str	    =	True

													if model_class in get_values(lowercase__) or not model_class.supports_gradient_checkpointing:
																			continue
													__UpperCAmelCase			:							Tuple	    =	model_class(lowercase__)
													model.to(lowercase__)
													model.gradient_checkpointing_enable()
													model.train()
													__UpperCAmelCase			:							Tuple	    =	self._prepare_for_class(lowercase__				,     lowercase__				,     return_labels=lowercase__)
													__UpperCAmelCase			:							Union[str, Any]	    =	model(**lowercase__).loss
													loss.backward()


	def  A(    self):
							__UpperCAmelCase    ,      __UpperCAmelCase			:							Optional[int]	    =	self.model_tester.prepare_config_and_inputs_for_common()

							__UpperCAmelCase			:							List[Any]	    =	_config_zero_init(lowercase__)
							for model_class in self.all_model_classes:
													__UpperCAmelCase			:							Tuple	    =	model_class(config=lowercase__)
													# Skip the check for the backbone
													__UpperCAmelCase			:							List[Any]	    =	[]
													for name, module in model.named_modules():
																			if module.__class__.__name__ == "DPTViTHybridEmbeddings":
																									__UpperCAmelCase			:							Optional[Any]	    =	[F"{name}.{key}" for key in module.state_dict().keys()]
																									break

													for name, param in model.named_parameters():
																			if param.requires_grad:
																									if name in backbone_params:
																															continue
																									self.assertIn(
																									    ((param.data.mean() * 1e9).round() / 1e9).item()				,     [0.0, 1.0]				,     msg=F"Parameter {name} of model {model_class} seems not properly initialized"				,     )


	@unittest.skip('''Will be fixed soon by reducing the size of the model used for common tests.''')
	def  A(    self):
							pass


	@slow
	def  A(    self):
							for model_name in DPT_PRETRAINED_MODEL_ARCHIVE_LIST[1:]:
													__UpperCAmelCase			:							Optional[int]	    =	DPTModel.from_pretrained(lowercase__)
													self.assertIsNotNone(lowercase__)


	def  A(    self):
							# We do this test only for DPTForDepthEstimation since it is the only model that uses readout_type
							__UpperCAmelCase    ,      __UpperCAmelCase			:							Tuple	    =	self.model_tester.prepare_config_and_inputs_for_common()
							__UpperCAmelCase			:							List[str]	    =	'''add'''
							with self.assertRaises(lowercase__):
													__UpperCAmelCase			:							Optional[Any]	    =	DPTForDepthEstimation(lowercase__)






def     __SCREAMING_SNAKE_CASE							(       )		->     Dict:




						'''simple docstring'''







						__UpperCAmelCase			:							str	    =	Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''      )
						return image





@require_torch
@require_vision
@slow
class 							lowerCamelCase   (  unittest.TestCase ):


	def  A(    self):
							__UpperCAmelCase			:							str	    =	DPTImageProcessor.from_pretrained('''Intel/dpt-hybrid-midas''')
							__UpperCAmelCase			:							str	    =	DPTForDepthEstimation.from_pretrained('''Intel/dpt-hybrid-midas''').to(lowercase__)

							__UpperCAmelCase			:							List[str]	    =	prepare_img()
							__UpperCAmelCase			:							Tuple	    =	image_processor(images=lowercase__				,     return_tensors='''pt''').to(lowercase__)

							# forward pass
							with torch.no_grad():
													__UpperCAmelCase			:							List[Any]	    =	model(**lowercase__)
													__UpperCAmelCase			:							str	    =	outputs.predicted_depth

							# verify the predicted depth
							__UpperCAmelCase			:							Union[str, Any]	    =	torch.Size((1, 3_8_4, 3_8_4))
							self.assertEqual(predicted_depth.shape				,     lowercase__)

							__UpperCAmelCase			:							List[str]	    =	torch.tensor(
							    [[[5.6_4_3_7, 5.6_1_4_6, 5.6_5_1_1], [5.4_3_7_1, 5.5_6_4_9, 5.5_9_5_8], [5.5_2_1_5, 5.5_1_8_4, 5.5_2_9_3]]]).to(lowercase__)

							self.assertTrue(torch.allclose(outputs.predicted_depth[:3, :3, :3] / 1_0_0				,     lowercase__				,     atol=1e-4))

