

from __future__ import annotations

import copy
import os
import tempfile
from importlib import import_module
from math import isnan

from transformers import is_tf_available
from transformers.models.auto import get_values
from transformers.testing_utils import _tf_gpu_memory_limit, require_tf, slow

from ..test_modeling_tf_common import ids_tensor


if is_tf_available():
				import numpy as np
				import tensorflow as tf

				from transformers import (
				    TF_MODEL_FOR_CAUSAL_LM_MAPPING,
				    TF_MODEL_FOR_IMAGE_CLASSIFICATION_MAPPING,
				    TF_MODEL_FOR_MASKED_LM_MAPPING,
				    TF_MODEL_FOR_MULTIPLE_CHOICE_MAPPING,
				    TF_MODEL_FOR_NEXT_SENTENCE_PREDICTION_MAPPING,
				    TF_MODEL_FOR_PRETRAINING_MAPPING,
				    TF_MODEL_FOR_QUESTION_ANSWERING_MAPPING,
				    TF_MODEL_FOR_SEQ_TO_SEQ_CAUSAL_LM_MAPPING,
				    TF_MODEL_FOR_SEQUENCE_CLASSIFICATION_MAPPING,
				    TF_MODEL_FOR_TOKEN_CLASSIFICATION_MAPPING,
				    TFSharedEmbeddings,
				)

				if _tf_gpu_memory_limit is not None:
								_a:      Tuple								=     tf.config.list_physical_devices("""GPU""")
								for gpu in gpus:
												# Restrict TensorFlow to only allocate x GB of memory on the GPUs
												try:
																tf.config.set_logical_device_configuration(
																    gpu, [tf.config.LogicalDeviceConfiguration(memory_limit=_tf_gpu_memory_limit)]
																)
																_a:      List[Any]								=     tf.config.list_logical_devices("""GPU""")
																print("""Logical GPUs""", logical_gpus)
												except RuntimeError as e:
																# Virtual devices must be set before GPUs have been initialized
																print(e)

@require_tf
class   __UpperCamelCase      :
	SCREAMING_SNAKE_CASE__      			=       None
	SCREAMING_SNAKE_CASE__      			=       ()
	SCREAMING_SNAKE_CASE__      			=       ()
	SCREAMING_SNAKE_CASE__      			=       True
	SCREAMING_SNAKE_CASE__      			=       True
	SCREAMING_SNAKE_CASE__      			=       True
	SCREAMING_SNAKE_CASE__      			=       False




	def  __A   (						self		:  List[Any]						,		lowerCAmelCase		:  List[Any]						,		lowerCAmelCase		:  Union[str, Any]						,		lowerCAmelCase		:  Union[str, Any]=False							):


		'''simple docstring'''







		UpperCAmelCase_    			=     copy.deepcopy(lowerCAmelCase							)

		if model_class in get_values(lowerCAmelCase							):
			UpperCAmelCase_    			=     {
			    k: tf.tile(tf.expand_dims(lowerCAmelCase						,		1							)						,		(1, self.model_tester.num_choices) + (1,) * (v.ndim - 1)							)
			    if isinstance(lowerCAmelCase						,		tf.Tensor							) and v.ndim > 0
			    else v
			    for k, v in inputs_dict.items()
			}

		if return_labels:
			if model_class in get_values(lowerCAmelCase							):
				UpperCAmelCase_    			=     tf.ones(self.model_tester.batch_size						,		dtype=tf.intaa							)
			elif model_class in get_values(lowerCAmelCase							):
				UpperCAmelCase_    			=     tf.zeros(self.model_tester.batch_size						,		dtype=tf.intaa							)
				UpperCAmelCase_    			=     tf.zeros(self.model_tester.batch_size						,		dtype=tf.intaa							)
			elif model_class in [
			    *get_values(lowerCAmelCase							),
			    *get_values(lowerCAmelCase							),
			]:
				UpperCAmelCase_    			=     tf.zeros(self.model_tester.batch_size						,		dtype=tf.intaa							)
			elif model_class in get_values(lowerCAmelCase							):
				UpperCAmelCase_    			=     tf.zeros(self.model_tester.batch_size						,		dtype=tf.intaa							)
			elif model_class in [
			    *get_values(lowerCAmelCase							),
			    *get_values(lowerCAmelCase							),
			    *get_values(lowerCAmelCase							),
			    *get_values(lowerCAmelCase							),
			    *get_values(lowerCAmelCase							),
			]:
				UpperCAmelCase_    			=     tf.zeros(
				    (self.model_tester.batch_size, self.model_tester.seq_length)						,		dtype=tf.intaa							)
		return inputs_dict




	@slow
	def  __A   (						self		:  int							):


		'''simple docstring'''







		UpperCAmelCase_			,				UpperCAmelCase_    			=     self.model_tester.prepare_config_and_inputs_for_common()
		for model_class in self.all_model_classes[:2]:
			UpperCAmelCase_    			=     self._prepare_for_class(lowerCAmelCase						,		lowerCAmelCase							)
			UpperCAmelCase_    			=     model_class(lowerCAmelCase							)

			@tf.function
			def run_in_graph_mode():
				return model(lowerCAmelCase							)

			UpperCAmelCase_    			=     run_in_graph_mode()
			self.assertIsNotNone(lowerCAmelCase							)




	@slow
	def  __A   (						self		:  Tuple							):


		'''simple docstring'''







		UpperCAmelCase_			,				UpperCAmelCase_    			=     self.model_tester.prepare_config_and_inputs_for_common()
		for model_class in self.all_model_classes[:2]:
			UpperCAmelCase_    			=     self._prepare_for_class(lowerCAmelCase						,		lowerCAmelCase							)
			UpperCAmelCase_    			=     model_class(lowerCAmelCase							)

			@tf.function(experimental_compile=lowerCAmelCase							)
			def run_in_graph_mode():
				return model(lowerCAmelCase							)

			UpperCAmelCase_    			=     run_in_graph_mode()
			self.assertIsNotNone(lowerCAmelCase							)




	@slow
	def  __A   (						self		:  Tuple							):


		'''simple docstring'''







		UpperCAmelCase_			,				UpperCAmelCase_    			=     self.model_tester.prepare_config_and_inputs_for_common()
		for model_class in self.all_model_classes[:2]:
			UpperCAmelCase_    			=     model_class(lowerCAmelCase							)
			if getattr(lowerCAmelCase						,		"hf_compute_loss"						,		lowerCAmelCase							):
				# Test that model correctly compute the loss with kwargs
				UpperCAmelCase_    			=     self._prepare_for_class(inputs_dict.copy()						,		lowerCAmelCase						,		return_labels=lowerCAmelCase							)
				# Is there a better way to remove these decoder inputs?
				UpperCAmelCase_    			=     {
				    key: val
				    for key, val in prepared_for_class.items()
				    if key not in ("head_mask", "decoder_head_mask", "cross_attn_head_mask", "decoder_input_ids")
				}

				UpperCAmelCase_    			=     {
				    "labels",
				    "label",
				    "label_ids",
				    "start_positions",
				    "start_position",
				    "end_positions",
				    "end_position",
				    "next_sentence_label",
				}
				UpperCAmelCase_    			=     possible_label_cols.intersection(set(lowerCAmelCase							)							)
				self.assertGreater(len(lowerCAmelCase							)						,		0						,		msg="No matching label names found!"							)
				UpperCAmelCase_    			=     {key: val for key, val in prepared_for_class.items() if key in label_names}
				UpperCAmelCase_    			=     {key: val for key, val in prepared_for_class.items() if key not in label_names}
				self.assertGreater(len(lowerCAmelCase							)						,		0							)

				# Make sure it works with XLA!
				model.compile(optimizer=tf.keras.optimizers.SGD(0.0							)						,		jit_compile=lowerCAmelCase							)
				# Make sure the model fits without crashing regardless of where we pass the labels
				UpperCAmelCase_    			=     model.fit(
				    lowerCAmelCase						,		validation_data=lowerCAmelCase						,		steps_per_epoch=1						,		validation_steps=1						,		shuffle=lowerCAmelCase						,		verbose=0						,		)
				UpperCAmelCase_    			=     history.history["loss"][0]
				self.assertTrue(not isnan(lowerCAmelCase							)							)
				UpperCAmelCase_    			=     history.history["val_loss"][0]
				self.assertTrue(not isnan(lowerCAmelCase							)							)

				# Now test it with separate labels, to make sure that path works in XLA too.
				UpperCAmelCase_    			=     model_class(lowerCAmelCase							)
				model.compile(optimizer=tf.keras.optimizers.SGD(0.0							)						,		jit_compile=lowerCAmelCase							)
				UpperCAmelCase_    			=     model.fit(
				    lowerCAmelCase						,		lowerCAmelCase						,		validation_data=(inputs_minus_labels, labels)						,		steps_per_epoch=1						,		validation_steps=1						,		shuffle=lowerCAmelCase						,		verbose=0						,		)

				UpperCAmelCase_    			=     history.history["loss"][0]
				self.assertTrue(not isnan(lowerCAmelCase							)							)
				UpperCAmelCase_    			=     history.history["val_loss"][0]
				self.assertTrue(not isnan(lowerCAmelCase							)							)




	@slow
	def  __A   (						self		:  Any							):


		'''simple docstring'''







		UpperCAmelCase_			,				UpperCAmelCase_    			=     self.model_tester.prepare_config_and_inputs_for_common()
		UpperCAmelCase_    			=     True
		UpperCAmelCase_    			=     True

		if hasattr(lowerCAmelCase						,		"use_cache"							):
			UpperCAmelCase_    			=     True

		UpperCAmelCase_    			=     getattr(self.model_tester						,		"encoder_seq_length"						,		self.model_tester.seq_length							)
		UpperCAmelCase_    			=     getattr(self.model_tester						,		"key_length"						,		lowerCAmelCase							)

		for model_class in self.all_model_classes[:2]:
			UpperCAmelCase_    			=     self._prepare_for_class(lowerCAmelCase						,		lowerCAmelCase							)
			UpperCAmelCase_    			=     model_class(lowerCAmelCase							)
			model.build()
			UpperCAmelCase_    			=     len(model(lowerCAmelCase							)							)

			for key in list(class_inputs_dict.keys()							):
				# Remove keys not in the serving signature, as the SavedModel will not be compiled to deal with them
				if key not in model.input_signature:
					del class_inputs_dict[key]
				# Check it's a tensor, in case the inputs dict has some bools in it too
				elif isinstance(class_inputs_dict[key]						,		tf.Tensor							) and class_inputs_dict[key].dtype.is_integer:
					UpperCAmelCase_    			=     tf.cast(class_inputs_dict[key]						,		tf.intaa							)

			if set(class_inputs_dict.keys()							) != set(model.input_signature.keys()							):
				continue  # Some models have inputs that the preparation functions don't create, we skip those

			with tempfile.TemporaryDirectory() as tmpdirname:
				model.save_pretrained(lowerCAmelCase						,		saved_model=lowerCAmelCase							)
				UpperCAmelCase_    			=     os.path.join(lowerCAmelCase						,		"saved_model"						,		"1"							)
				UpperCAmelCase_    			=     tf.keras.models.load_model(lowerCAmelCase							)
				UpperCAmelCase_    			=     model(lowerCAmelCase							)

				if self.is_encoder_decoder:
					UpperCAmelCase_    			=     outputs["encoder_hidden_states"]
					UpperCAmelCase_    			=     outputs["encoder_attentions"]
				else:
					UpperCAmelCase_    			=     outputs["hidden_states"]
					UpperCAmelCase_    			=     outputs["attentions"]

				self.assertEqual(len(lowerCAmelCase							)						,		lowerCAmelCase							)

				UpperCAmelCase_    			=     getattr(
				    self.model_tester						,		"expected_num_hidden_layers"						,		self.model_tester.num_hidden_layers + 1							)

				self.assertEqual(len(lowerCAmelCase							)						,		lowerCAmelCase							)
				self.assertListEqual(
				    list(output_hidden_states[0].shape[-2:]							)						,		[self.model_tester.seq_length, self.model_tester.hidden_size]						,		)

				self.assertEqual(len(lowerCAmelCase							)						,		self.model_tester.num_hidden_layers							)
				self.assertListEqual(
				    list(output_attentions[0].shape[-3:]							)						,		[self.model_tester.num_attention_heads, encoder_seq_length, encoder_key_length]						,		)




	@slow
	def  __A   (						self		:  List[Any]							):


		'''simple docstring'''







		tf.keras.mixed_precision.set_global_policy("mixed_float16"							)

		# try/finally block to ensure subsequent tests run in float32
		try:
			UpperCAmelCase_			,				UpperCAmelCase_    			=     self.model_tester.prepare_config_and_inputs_for_common()
			for model_class in self.all_model_classes[:2]:
				UpperCAmelCase_    			=     self._prepare_for_class(lowerCAmelCase						,		lowerCAmelCase							)
				UpperCAmelCase_    			=     model_class(lowerCAmelCase							)
				UpperCAmelCase_    			=     model(lowerCAmelCase							)

				self.assertIsNotNone(lowerCAmelCase							)
		finally:
			tf.keras.mixed_precision.set_global_policy("float32"							)




	@slow
	def  __A   (						self		:  List[str]							):


		'''simple docstring'''







		UpperCAmelCase_			,				UpperCAmelCase_    			=     self.model_tester.prepare_config_and_inputs_for_common()
		# head_mask and decoder_head_mask has different shapes than other input args
		if "head_mask" in inputs_dict:
			del inputs_dict["head_mask"]
		if "decoder_head_mask" in inputs_dict:
			del inputs_dict["decoder_head_mask"]
		if "cross_attn_head_mask" in inputs_dict:
			del inputs_dict["cross_attn_head_mask"]
		UpperCAmelCase_    			=     {
		    module_member
		    for model_class in self.all_model_classes
		    for module in (import_module(model_class.__module__							),)
		    for module_member_name in dir(lowerCAmelCase							)
		    if module_member_name.endswith("MainLayer"							)
		    for module_member in (getattr(lowerCAmelCase						,		lowerCAmelCase							),)
		    if isinstance(lowerCAmelCase						,		lowerCAmelCase							)
		    and tf.keras.layers.Layer in module_member.__bases__
		    and getattr(lowerCAmelCase						,		"_keras_serializable"						,		lowerCAmelCase							)
		}

		for main_layer_class in tf_main_layer_classes:
			# T5MainLayer needs an embed_tokens parameter when called without the inputs_embeds parameter
			if "T5" in main_layer_class.__name__:
				# Take the same values than in TFT5ModelTester for this shared layer
				UpperCAmelCase_    			=     TFSharedEmbeddings(self.model_tester.vocab_size						,		self.model_tester.hidden_size						,		name="shared"							)
				UpperCAmelCase_    			=     False
				UpperCAmelCase_    			=     main_layer_class(lowerCAmelCase						,		embed_tokens=lowerCAmelCase							)
			else:
				UpperCAmelCase_    			=     main_layer_class(lowerCAmelCase							)

			UpperCAmelCase_    			=     {
			    name: tf.keras.Input(tensor.shape[1:]						,		dtype=tensor.dtype							) for name, tensor in inputs_dict.items()
			}

			if hasattr(self.model_tester						,		"num_labels"							):
				UpperCAmelCase_    			=     self.model_tester.num_labels
			else:
				UpperCAmelCase_    			=     2

			UpperCAmelCase_    			=     tf.data.Dataset.from_tensor_slices(
			    (inputs_dict, np.ones((self.model_tester.batch_size, self.model_tester.seq_length, num_labels, 1)							))							).batch(1							)

			UpperCAmelCase_    			=     main_layer(lowerCAmelCase							)[0]
			UpperCAmelCase_    			=     tf.keras.layers.Dense(lowerCAmelCase						,		activation="softmax"						,		name="outputs"							)(lowerCAmelCase							)
			UpperCAmelCase_    			=     tf.keras.models.Model(inputs=lowerCAmelCase						,		outputs=[outputs]							)

			model.compile(loss="binary_crossentropy"						,		optimizer="adam"						,		metrics=["binary_accuracy"]							)
			model.fit(lowerCAmelCase						,		epochs=1							)

			with tempfile.TemporaryDirectory() as tmpdirname:
				UpperCAmelCase_    			=     os.path.join(lowerCAmelCase						,		"keras_model.h5"							)
				model.save(lowerCAmelCase							)
				if "T5" in main_layer_class.__name__:
					UpperCAmelCase_    			=     tf.keras.models.load_model(
					    lowerCAmelCase						,		custom_objects={
					        main_layer_class.__name__: main_layer_class,
					        "TFSharedEmbeddings": TFSharedEmbeddings,
					    }						,		)
				else:
					UpperCAmelCase_    			=     tf.keras.models.load_model(
					    lowerCAmelCase						,		custom_objects={main_layer_class.__name__: main_layer_class}							)
				assert isinstance(lowerCAmelCase						,		tf.keras.Model							)
				model(lowerCAmelCase							)




	@slow
	def  __A   (						self		:  Union[str, Any]							):


		'''simple docstring'''







		UpperCAmelCase_			,				UpperCAmelCase_    			=     self.model_tester.prepare_config_and_inputs_for_common()

		for model_class in self.all_model_classes[:2]:
			UpperCAmelCase_    			=     model_class(lowerCAmelCase							)

			UpperCAmelCase_    			=     copy.deepcopy(lowerCAmelCase							)

			if not self.is_encoder_decoder:
				UpperCAmelCase_    			=     inputs["input_ids"]
				del inputs["input_ids"]
			else:
				UpperCAmelCase_    			=     inputs["input_ids"]
				UpperCAmelCase_    			=     inputs.get("decoder_input_ids"						,		lowerCAmelCase							)
				del inputs["input_ids"]
				inputs.pop("decoder_input_ids"						,		lowerCAmelCase							)

			if not self.is_encoder_decoder:
				UpperCAmelCase_    			=     model.get_input_embeddings()(lowerCAmelCase							)
			else:
				UpperCAmelCase_    			=     model.get_input_embeddings()(lowerCAmelCase							)
				UpperCAmelCase_    			=     model.get_input_embeddings()(lowerCAmelCase							)

			UpperCAmelCase_    			=     self._prepare_for_class(lowerCAmelCase						,		lowerCAmelCase							)

			@tf.function
			def run_in_graph_mode():
				return model(lowerCAmelCase							)

			UpperCAmelCase_    			=     run_in_graph_mode()
			self.assertIsNotNone(lowerCAmelCase							)




	def  __A   (						self		:  str						,		lowerCAmelCase		:  str						,		lowerCAmelCase		:  Any							):


		'''simple docstring'''







		UpperCAmelCase_    			=     []
		if model.config.bos_token_id is not None:
			special_tokens.append(model.config.bos_token_id							)
		if model.config.pad_token_id is not None:
			special_tokens.append(model.config.pad_token_id							)
		if model.config.eos_token_id is not None:
			special_tokens.append(model.config.eos_token_id							)

		# create random bad tokens that are not special tokens
		UpperCAmelCase_    			=     []
		while len(lowerCAmelCase							) < num_bad_tokens:
			UpperCAmelCase_    			=     tf.squeeze(ids_tensor((1, 1)						,		self.model_tester.vocab_size							)						,		0							).numpy()[0]
			if token not in special_tokens:
				bad_tokens.append(lowerCAmelCase							)
		return bad_tokens




	def  __A   (						self		:  Optional[Any]						,		lowerCAmelCase		:  Optional[Any]							):


		'''simple docstring'''







		for token_id in output_ids[0].numpy().tolist():
			self.assertGreaterEqual(lowerCAmelCase						,		0							)
			self.assertLess(lowerCAmelCase						,		self.model_tester.vocab_size							)




	def  __A   (						self		:  Union[str, Any]						,		lowerCAmelCase		:  int						,		lowerCAmelCase		:  List[Any]							):


		'''simple docstring'''







		for bad_word_ids in bad_words_ids:
			# for all slices in batch
			for generated_ids_slice in generated_ids:
				# for all word idx
				for i in range(len(lowerCAmelCase							)						,		len(lowerCAmelCase							)							):
					# if tokens match
					if generated_ids_slice[i - len(lowerCAmelCase							) : i] == bad_word_ids:
						return True
		return False


import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
				import torch_xla.core.xla_model as xm  # noqa: F401

				_a:      List[Any]								=     True
except ImportError:
				_a:      int								=     False


# Cache this result has it's a C FFI call which can be pretty time-consuming
_a:      str								=     torch.distributed.is_available()



def  __lowerCAmelCase (  A      ):
	# Check we're not importing a "pkg_name" directory somewhere but the actual library by trying to grab the version
	UpperCAmelCase_    			=     importlib.util.find_spec(A      ) is not None
	if package_exists:
		try:
			UpperCAmelCase_    			=     importlib.metadata.metadata(A      )
			return True
		except importlib.metadata.PackageNotFoundError:
			return False



def  __lowerCAmelCase (  ):
	return _torch_distributed_available



def  __lowerCAmelCase (  ):
	try:
		pass
	except ImportError:
		print(
		    "Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not"
		    " detected. If you see \"ValueError: Invalid backend: 'ccl'\" error, please install Intel(R) oneCCL"
		    " Bindings for PyTorch*."      )
	return (
	    importlib.util.find_spec("torch_ccl"      ) is not None
	    or importlib.util.find_spec("oneccl_bindings_for_pytorch"      ) is not None
	)



def  __lowerCAmelCase (  ):
	return importlib.metadata.version("oneccl_bind_pt"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("transformer_engine"      )



@lru_cache
def  __lowerCAmelCase (  A=True      ):
	# Due to bugs on the amp series GPUs, we disable torch-xla on them
	if torch.cuda.is_available():
		return False
	if _tpu_available and check_device:
		try:
			# Will raise a RuntimeError if no XLA configuration is found
			UpperCAmelCase_    			=     xm.xla_device()
			return True
		except RuntimeError:
			return False
	return _tpu_available



def  __lowerCAmelCase (  ):
	return _is_package_available("deepspeed"      )



def  __lowerCAmelCase (  A=False      ):
	if is_tpu_available():
		return not ignore_tpu
	if torch.cuda.is_available():
		return torch.cuda.is_bfaa_supported()
	if is_npu_available():
		return False
	return True



def  __lowerCAmelCase (  ):
	UpperCAmelCase_    			=     _is_package_available("bitsandbytes"      )
	if package_exists:
		UpperCAmelCase_    			=     version.parse(importlib.metadata.version("bitsandbytes"      )      )
		return compare_versions(A    ,       ">="    ,       "0.39.0"      )
	return False



def  __lowerCAmelCase (  ):
	UpperCAmelCase_    			=     _is_package_available("bitsandbytes"      )
	if package_exists:
		UpperCAmelCase_    			=     version.parse(importlib.metadata.version("bitsandbytes"      )      )
		return compare_versions(A    ,       ">="    ,       "0.37.2"      )
	return False



def  __lowerCAmelCase (  ):
	return _is_package_available("bitsandbytes"      )



def  __lowerCAmelCase (  ):
	if strtobool(os.environ.get("ACCELERATE_USE_MEGATRON_LM"    ,       "False"      )      ) == 1:
		UpperCAmelCase_    			=     importlib.util.find_spec("megatron"      ) is not None
		if package_exists:
			try:
				UpperCAmelCase_    			=     parse(importlib.metadata.version("megatron-lm"      )      )
				return compare_versions(A    ,       ">="    ,       "2.2.0"      )
			except Exception as e:
				warnings.warn(F"Parse Megatron version failed. Exception:{e}"      )
				return False



def  __lowerCAmelCase (  ):
	return _is_package_available("safetensors"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("transformers"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("datasets"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("aim"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("tensorboard"      ) or _is_package_available("tensorboardX"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("wandb"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("comet_ml"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("boto3"      )



def  __lowerCAmelCase (  ):
	if _is_package_available("rich"      ):
		if "ACCELERATE_DISABLE_RICH" in os.environ:
			warnings.warn(
			    "`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`."      )
			return not parse_flag_from_env("ACCELERATE_DISABLE_RICH"    ,       A      )
		return parse_flag_from_env("ACCELERATE_ENABLE_RICH"    ,       A      )
	return False



def  __lowerCAmelCase (  ):
	return _is_package_available("sagemaker"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("tqdm"      )



def  __lowerCAmelCase (  ):
	return _is_package_available("mlflow"      )



def  __lowerCAmelCase (  ):
	return is_torch_version(">="    ,       "1.12"      ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()



def  __lowerCAmelCase (  ):
	def get_major_and_minor_from_version(A      ):
		return str(version.parse(A      ).major      ) + "." + str(version.parse(A      ).minor      )

	UpperCAmelCase_    			=     importlib.metadata.version("torch"      )
	if importlib.util.find_spec("intel_extension_for_pytorch"      ) is None:
		return False
	UpperCAmelCase_    			=     "N/A"
	try:
		UpperCAmelCase_    			=     importlib.metadata.version("intel_extension_for_pytorch"      )
	except importlib.metadata.PackageNotFoundError:
		return False
	UpperCAmelCase_    			=     get_major_and_minor_from_version(_torch_version      )
	UpperCAmelCase_    			=     get_major_and_minor_from_version(_ipex_version      )
	if torch_major_and_minor != ipex_major_and_minor:
		warnings.warn(
		    F"Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,"
		    F" but PyTorch {_torch_version} is found. Please switch to the matching version and run again."      )
		return False
	return True



@lru_cache
def  __lowerCAmelCase (  A=False      ):
	if importlib.util.find_spec("torch"      ) is None or importlib.util.find_spec("torch_npu"      ) is None:
		return False

	import torch
	import torch_npu  # noqa: F401

	if check_device:
		try:
			# Will raise a RuntimeError if no NPU is found
			UpperCAmelCase_    			=     torch.npu.device_count()
			return torch.npu.is_available()
		except RuntimeError:
			return False
	return hasattr(A    ,       "npu"      ) and torch.npu.is_available()




@lru_cache
def  __lowerCAmelCase (  A=False      ):
	if not parse_flag_from_env("ACCELERATE_USE_XPU"    ,       default=A      ):
		return False
	if is_ipex_available():
		import torch

		if is_torch_version("<="    ,       "1.12"      ):
			return False
	else:
		return False

	import intel_extension_for_pytorch  # noqa: F401

	if check_device:
		try:
			# Will raise a RuntimeError if no XPU  is found
			UpperCAmelCase_    			=     torch.xpu.device_count()
			return torch.xpu.is_available()
		except RuntimeError:
			return False
	return hasattr(A    ,       "xpu"      ) and torch.xpu.is_available()