






import copy
from typing import Dict, List, Optional

from ...configuration_utils import PretrainedConfig
from ...utils import logging
from ..auto import CONFIG_MAPPING


__lowerCamelCase :      Optional[int]										=				{
    """facebook/mask2former-swin-small-coco-instance""": (
        """https://huggingface.co/facebook/mask2former-swin-small-coco-instance/blob/main/config.json"""
    )
    # See all Mask2Former models at https://huggingface.co/models?filter=mask2former
}

__lowerCamelCase :      Any										=				logging.get_logger(__name__)



class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					a_     =		"mask2former"
					a_     =		["swin"]
					a_     =		{"hidden_size": "hidden_dim"}



					def __init__( self					:  Union[str, Any]							,		__A					:  Optional[Dict] = None							,		__A					:  int = 2_5_6							,		__A					:  int = 2_5_6							,		__A					:  int = 2_5_6							,		__A					:  int = 1_0_2_4							,		__A					:  str = "relu"							,		__A					:  int = 6							,		__A					:  int = 1_0							,		__A					:  int = 8							,		__A					:  float = 0.0							,		__A					:  int = 2_0_4_8							,		__A					:  bool = False							,		__A					:  bool = False							,		__A					:  int = 4							,		__A					:  int = 2_5_5							,		__A					:  int = 1_0_0							,		__A					:  float = 0.1							,		__A					:  float = 2.0							,		__A					:  float = 5.0							,		__A					:  float = 5.0							,		__A					:  int = 1_2_5_4_4							,		__A					:  float = 3.0							,		__A					:  float = 0.7_5							,		__A					:  float = 0.0_2							,		__A					:  float = 1.0							,		__A					:  bool = True							,		__A					:  List[int] = [4, 8, 1_6, 3_2]							,		__A					:  bool = None							,		**__A					:  str							,		):
						if backbone_config is None:
							logger.info("`backbone_config` is `None`. Initializing the config with the default `Swin` backbone."	)
							snake_case__			:				int       =  CONFIG_MAPPING["swin"](
							    image_size=2_2_4							,		in_channels=3							,		patch_size=4							,		embed_dim=9_6							,		depths=[2, 2, 1_8, 2]							,		num_heads=[3, 6, 1_2, 2_4]							,		window_size=7							,		drop_path_rate=0.3							,		use_absolute_embeddings=__A							,		out_features=["stage1", "stage2", "stage3", "stage4"]							,		)

						if isinstance(__A							,		__A	):
							snake_case__			:				Optional[int]       =  backbone_config.pop("model_type"	)
							snake_case__			:				Optional[Any]       =  CONFIG_MAPPING[backbone_model_type]
							snake_case__			:				int       =  config_class.from_dict(__A	)

						# verify that the backbone is supported
						if backbone_config.model_type not in self.backbones_supported:
							logger.warning_once(
							    f'''Backbone {backbone_config.model_type} is not a supported model and may not be compatible with Mask2Former. '''
							    f'''Supported model types: {','.join(self.backbones_supported	)}'''	)

						snake_case__			:				Any       =  backbone_config
						snake_case__			:				Any       =  feature_size
						snake_case__			:				List[str]       =  mask_feature_size
						snake_case__			:				Optional[Any]       =  hidden_dim
						snake_case__			:				Optional[int]       =  encoder_feedforward_dim
						snake_case__			:				List[str]       =  activation_function
						snake_case__			:				Any       =  encoder_layers
						snake_case__			:				List[Any]       =  decoder_layers
						snake_case__			:				List[str]       =  num_attention_heads
						snake_case__			:				Union[str, Any]       =  dropout
						snake_case__			:				Dict       =  dim_feedforward
						snake_case__			:				Dict       =  pre_norm
						snake_case__			:				int       =  enforce_input_projection
						snake_case__			:				Union[str, Any]       =  common_stride
						snake_case__			:				Union[str, Any]       =  ignore_value
						snake_case__			:				List[Any]       =  num_queries
						snake_case__			:				str       =  no_object_weight
						snake_case__			:				List[Any]       =  class_weight
						snake_case__			:				Optional[Any]       =  mask_weight
						snake_case__			:				Any       =  dice_weight
						snake_case__			:				str       =  train_num_points
						snake_case__			:				Union[str, Any]       =  oversample_ratio
						snake_case__			:				Tuple       =  importance_sample_ratio
						snake_case__			:				Optional[Any]       =  init_std
						snake_case__			:				List[Any]       =  init_xavier_std
						snake_case__			:				int       =  use_auxiliary_loss
						snake_case__			:				Any       =  feature_strides
						snake_case__			:				Union[str, Any]       =  output_auxiliary_logits
						snake_case__			:				Optional[int]       =  decoder_layers

						super().__init__(**__A	)



					@classmethod
					def 	_lowercase					( cls					:  str							,		__A					:  PretrainedConfig							,		**__A					:  Dict	):
						return cls(
						    backbone_config=__A							,		**__A							,		)






					def 	_lowercase					( self					:  Union[str, Any]	):
						snake_case__			:				Dict       =  copy.deepcopy(self.__dict__	)
						snake_case__			:				Any       =  self.backbone_config.to_dict()
						snake_case__			:				Optional[int]       =  self.__class__.model_type
						return output











import random
import timeit
from functools import wraps
from typing import Callable, Optional

from ..configuration_utils import PretrainedConfig
from ..models.auto.modeling_tf_auto import TF_MODEL_MAPPING, TF_MODEL_WITH_LM_HEAD_MAPPING
from ..utils import is_pyanvml_available, is_tf_available, logging
from .benchmark_utils import (
    Benchmark,
    Memory,
    MemorySummary,
    measure_peak_memory_cpu,
    start_memory_tracing,
    stop_memory_tracing,
)


if is_tf_available():
	import tensorflow as tf
	from tensorflow.python.framework.errors_impl import ResourceExhaustedError

	from .benchmark_args_tf import TensorFlowBenchmarkArguments

if is_pyanvml_available():
	import pyanvml.pyanvml as nvml

__lowerCamelCase :      List[Any]										=				logging.get_logger(__name__)







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							bool					,  snake_case_ :							bool       ):
	def run_func(snake_case_ :							str       ):
		@wraps(snake_case_       )
		def run_in_eager_mode(*snake_case_ :							str					,  **snake_case_ :							Union[str, Any]       ):
			return func(*snake_case_					,  **snake_case_       )

		@wraps(snake_case_       )
		@tf.function(experimental_compile=snake_case_       )
		def run_in_graph_mode(*snake_case_ :							List[Any]					,  **snake_case_ :							List[Any]       ):
			return func(*snake_case_					,  **snake_case_       )

		if do_eager_mode is True:
			if use_xla is not False:
				raise ValueError(
				    "Cannot run model in XLA, if `args.eager_mode` is set to `True`. Please set `args.eager_mode=False`."       )
			return run_in_eager_mode
		else:
			return run_in_graph_mode

	return run_func







def 					SCREAMING_SNAKE_CASE    (			snake_case_ :							int					,  snake_case_ :							int					,  snake_case_ :							int       ):
	snake_case__			:				Dict       =  random.Random()
	snake_case__			:				List[Any]       =  [rng.randint(0					,  vocab_size - 1       ) for i in range(batch_size * sequence_length       )]
	return tf.constant(snake_case_					,  shape=(batch_size, sequence_length)					,  dtype=tf.intaa       )





class 							SCREAMING_SNAKE_CASE__						(    UpperCamelCase_						):





					"""simple docstring"""






					a_     =		42
					a_     =		42
					a_     =		"TensorFlow"



					@property
					def 	_lowercase					( self					:  List[str]	):
						return tf.__version__



					def 	_lowercase					( self					:  List[str]							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						# initialize GPU on separate process
						snake_case__			:				str       =  self.args.strategy
						if strategy is None:
							raise ValueError("A device strategy has to be initialized before using TensorFlow."	)
						snake_case__			:				Dict       =  self._prepare_inference_func(__A							,		__A							,		__A	)
						return self._measure_speed(_inference	)



					def 	_lowercase					( self					:  Tuple							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						snake_case__			:				Optional[int]       =  self.args.strategy
						if strategy is None:
							raise ValueError("A device strategy has to be initialized before using TensorFlow."	)
						snake_case__			:				Any       =  self._prepare_train_func(__A							,		__A							,		__A	)
						return self._measure_speed(_train	)



					def 	_lowercase					( self					:  List[Any]							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						# initialize GPU on separate process
						if self.args.is_gpu:
							tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx]							,		__A	)
						snake_case__			:				List[str]       =  self.args.strategy
						if strategy is None:
							raise ValueError("A device strategy has to be initialized before using TensorFlow."	)
						snake_case__			:				Optional[Any]       =  self._prepare_inference_func(__A							,		__A							,		__A	)
						return self._measure_memory(_inference	)



					def 	_lowercase					( self					:  str							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						if self.args.is_gpu:
							tf.config.experimental.set_memory_growth(self.args.gpu_list[self.args.device_idx]							,		__A	)
						snake_case__			:				List[Any]       =  self.args.strategy
						if strategy is None:
							raise ValueError("A device strategy has to be initialized before using TensorFlow."	)

						snake_case__			:				int       =  self._prepare_train_func(__A							,		__A							,		__A	)
						return self._measure_memory(_train	)



					def 	_lowercase					( self					:  Union[str, Any]							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						snake_case__			:				int       =  self.config_dict[model_name]

						if self.args.fpaa:
							raise NotImplementedError("Mixed precision is currently not supported."	)

						snake_case__			:				Tuple       =  (
						    hasattr(__A							,		"architectures"	)
						    and isinstance(config.architectures							,		__A	)
						    and len(config.architectures	) > 0
						)
						if not self.args.only_pretrain_model and has_model_class_in_config:
							try:
								snake_case__			:				Dict       =  "TF" + config.architectures[0]  # prepend 'TF' for tensorflow model
								snake_case__			:				Union[str, Any]       =  __import__("transformers"							,		fromlist=[model_class]	)
								snake_case__			:				Any       =  getattr(__A							,		__A	)
								snake_case__			:				Dict       =  model_cls(__A	)
							except ImportError:
								raise ImportError(
								    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
								    " set `--only_pretrain_model` or `args.only_pretrain_model=True`."	)
						else:
							snake_case__			:				Dict       =  TF_MODEL_MAPPING[config.__class__](__A	)

						# encoder-decoder has vocab size saved differently
						snake_case__			:				Optional[int]       =  config.vocab_size if hasattr(__A							,		"vocab_size"	) else config.encoder.vocab_size
						snake_case__			:				List[Any]       =  random_input_ids(__A							,		__A							,		__A	)

						@run_with_tf_optimizations(self.args.eager_mode							,		self.args.use_xla	)
						def encoder_decoder_forward():
							return model(__A							,		decoder_input_ids=__A							,		training=__A	)

						@run_with_tf_optimizations(self.args.eager_mode							,		self.args.use_xla	)
						def encoder_forward():
							return model(__A							,		training=__A	)

						snake_case__			:				Optional[int]       =  encoder_decoder_forward if config.is_encoder_decoder else encoder_forward

						return _inference



					def 	_lowercase					( self					:  List[str]							,		__A					:  str							,		__A					:  int							,		__A					:  int	):
						snake_case__			:				Optional[Any]       =  self.config_dict[model_name]

						if self.args.eager_mode is not False:
							raise ValueError("Training cannot be done in eager mode. Please make sure that `args.eager_mode = False`."	)

						if self.args.fpaa:
							raise NotImplementedError("Mixed precision is currently not supported."	)

						snake_case__			:				Any       =  (
						    hasattr(__A							,		"architectures"	)
						    and isinstance(config.architectures							,		__A	)
						    and len(config.architectures	) > 0
						)
						if not self.args.only_pretrain_model and has_model_class_in_config:
							try:
								snake_case__			:				Dict       =  "TF" + config.architectures[0]  # prepend 'TF' for tensorflow model
								snake_case__			:				List[Any]       =  __import__("transformers"							,		fromlist=[model_class]	)
								snake_case__			:				Optional[int]       =  getattr(__A							,		__A	)
								snake_case__			:				str       =  model_cls(__A	)
							except ImportError:
								raise ImportError(
								    f'''{model_class} does not exist. If you just want to test the pretrained model, you might want to'''
								    " set `--only_pretrain_model` or `args.only_pretrain_model=True`."	)
						else:
							snake_case__			:				Union[str, Any]       =  TF_MODEL_WITH_LM_HEAD_MAPPING[config.__class__](__A	)

						# encoder-decoder has vocab size saved differently
						snake_case__			:				Union[str, Any]       =  config.vocab_size if hasattr(__A							,		"vocab_size"	) else config.encoder.vocab_size
						snake_case__			:				List[str]       =  random_input_ids(__A							,		__A							,		__A	)

						@run_with_tf_optimizations(self.args.eager_mode							,		self.args.use_xla	)
						def encoder_decoder_train():
							snake_case__			:				str       =  model(__A							,		decoder_input_ids=__A							,		labels=__A							,		training=__A	)[0]
							snake_case__			:				Dict       =  tf.gradients(__A							,		model.trainable_variables	)
							return gradients

						@run_with_tf_optimizations(self.args.eager_mode							,		self.args.use_xla	)
						def encoder_train():
							snake_case__			:				Optional[Any]       =  model(__A							,		labels=__A							,		training=__A	)[0]
							snake_case__			:				Dict       =  tf.gradients(__A							,		model.trainable_variables	)
							return gradients

						snake_case__			:				int       =  encoder_decoder_train if config.is_encoder_decoder else encoder_train

						return _train



					def 	_lowercase					( self					:  int							,		__A					:  List[Any]	):
						with self.args.strategy.scope():
							try:
								if self.args.is_tpu or self.args.use_xla:
									# run additional 10 times to stabilize compilation for tpu
									logger.info("Do inference on TPU. Running model 5 times to stabilize compilation"	)
									timeit.repeat(__A							,		repeat=1							,		number=5	)

								# as written in https://docs.python.org/2/library/timeit.html#timeit.Timer.repeat, min should be taken rather than the average
								snake_case__			:				Optional[Any]       =  timeit.repeat(
								    __A							,		repeat=self.args.repeat							,		number=1_0							,		)

								return min(__A	) / 1_0.0
							except ResourceExhaustedError as e:
								self.print_fn(f'''Doesn\'t fit on GPU. {e}'''	)






					def 	_lowercase					( self					:  str							,		__A					:  Callable[[], None]	):
						logger.info(
						    "Note that TensorFlow allocates more memory than "
						    "it might need to speed up computation. "
						    "The memory reported here corresponds to the memory "
						    "reported by `nvidia-smi`, which can vary depending "
						    "on total available memory on the GPU that is used."	)
						with self.args.strategy.scope():
							try:
								if self.args.trace_memory_line_by_line:
									if not self.args.eager_mode:
										raise ValueError(
										    "`args.eager_mode` is set to `False`. Make sure to run model in eager mode to measure memory"
										    " consumption line by line."	)
									snake_case__			:				Optional[int]       =  start_memory_tracing("transformers"	)

								if self.args.is_tpu:
									# tpu
									raise NotImplementedError(
									    "Memory Benchmarking is currently not implemented for TPU. Please disable memory benchmarking"
									    " with `args.memory=False`"	)
								elif self.args.is_gpu:
									# gpu
									if not is_pyanvml_available():
										logger.warning(
										    "py3nvml not installed, we won't log GPU memory usage. "
										    "Install py3nvml (pip install py3nvml) to log information about GPU."	)
										snake_case__			:				List[str]       =  "N/A"
									else:
										logger.info(
										    "Measuring total GPU usage on GPU device. Make sure to not have additional processes"
										    " running on the same GPU."	)
										# init nvml
										nvml.nvmlInit()
										func()
										snake_case__			:				Any       =  nvml.nvmlDeviceGetHandleByIndex(self.args.device_idx	)
										snake_case__			:				Optional[Any]       =  nvml.nvmlDeviceGetMemoryInfo(__A	)
										snake_case__			:				Optional[int]       =  meminfo.used
										snake_case__			:				Any       =  Memory(__A	)
										# shutdown nvml
										nvml.nvmlShutdown()
								else:
									# cpu
									if self.args.trace_memory_line_by_line:
										logger.info(
										    "When enabling line by line tracing, the max peak memory for CPU is inaccurate in"
										    " TensorFlow."	)
										snake_case__			:				int       =  None
									else:
										snake_case__			:				Any       =  measure_peak_memory_cpu(__A	)
										snake_case__			:				Tuple       =  Memory(__A	) if isinstance(__A							,		__A	) else memory_bytes
								if self.args.trace_memory_line_by_line:
									snake_case__			:				Optional[int]       =  stop_memory_tracing(__A	)
									if memory is None:
										snake_case__			:				Dict       =  summary.total
								else:
									snake_case__			:				List[str]       =  None

								return memory, summary
							except ResourceExhaustedError as e:
								self.print_fn(f'''Doesn\'t fit on GPU. {e}'''	)
								return "N/A", None



