




'''simple docstring'''







import warnings
from dataclasses import dataclass, field
from typing import Optional, Tuple

from .training_args import TrainingArguments
from .utils import cached_property, is_tf_available, logging, requires_backends


_lowerCamelCase							:						Optional[int]								=  logging.get_logger(__name__)

if is_tf_available():
						import tensorflow as tf





@dataclass
class      lowerCamelCase_							(       __UpperCAmelCase							):




					'''simple docstring'''






					lowercase__    = """tf"""
					lowercase__    = field(
					    default=__UpperCAmelCase,       metadata={"""help""": """Name of TPU"""},       )

					lowercase__    = field(
					    default=__UpperCAmelCase,       metadata={"""help""": """Zone of TPU"""},       )

					lowercase__    = field(
					    default=__UpperCAmelCase,       metadata={"""help""": """Name of Cloud TPU-enabled project"""},       )

					lowercase__    = field(
					    default=1.0,       metadata={"""help""": """Power for the Polynomial decay LR scheduler."""},       )

					lowercase__    = field(default=__UpperCAmelCase,       metadata={"""help""": """Whether to activate the XLA compilation or not"""}							)






					@cached_property
					def 				__SCREAMING_SNAKE_CASE	(    self ):
							requires_backends(self,  ["tf"] )
							logger.info("Tensorflow: setting up strategy" )

							_snake_case	       =       tf.config.list_physical_devices("GPU" )

							# Set to float16 at first
							if self.fpaa:
									tf.keras.mixed_precision.set_global_policy("mixed_float16" )

							if self.no_cuda:
									_snake_case	       =       tf.distribute.OneDeviceStrategy(device="/cpu:0" )
							else:
									try:
											if self.tpu_name:
													_snake_case	       =       tf.distribute.cluster_resolver.TPUClusterResolver(
													    self.tpu_name,  zone=self.tpu_zone,  project=self.gcp_project )
											else:
													_snake_case	       =       tf.distribute.cluster_resolver.TPUClusterResolver()
									except ValueError:
											if self.tpu_name:
													raise RuntimeError(f'''Couldn\'t connect to TPU {self.tpu_name}!''' )
											else:
													_snake_case	       =       None

									if tpu:
											# Set to bfloat16 in case of TPU
											if self.fpaa:
													tf.keras.mixed_precision.set_global_policy("mixed_bfloat16" )

											tf.config.experimental_connect_to_cluster(UpperCAmelCase_ )
											tf.tpu.experimental.initialize_tpu_system(UpperCAmelCase_ )

											_snake_case	       =       tf.distribute.TPUStrategy(UpperCAmelCase_ )

									elif len(UpperCAmelCase_ ) == 0:
											_snake_case	       =       tf.distribute.OneDeviceStrategy(device="/cpu:0" )
									elif len(UpperCAmelCase_ ) == 1:
											_snake_case	       =       tf.distribute.OneDeviceStrategy(device="/gpu:0" )
									elif len(UpperCAmelCase_ ) > 1:
											# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
											_snake_case	       =       tf.distribute.MirroredStrategy()
									else:
											raise ValueError("Cannot find the proper strategy, please check your environment properties." )

							return strategy






					@property
					def 				__SCREAMING_SNAKE_CASE	(    self ):
							requires_backends(self,  ["tf"] )
							return self._setup_strategy






					@property
					def 				__SCREAMING_SNAKE_CASE	(    self ):
							requires_backends(self,  ["tf"] )
							return self._setup_strategy.num_replicas_in_sync






					@property
					def 				__SCREAMING_SNAKE_CASE	(    self ):
							return False  # TF Logging is handled by Keras not the Trainer






					@property
					def 				__SCREAMING_SNAKE_CASE	(    self ):
							if self.per_gpu_train_batch_size:
									logger.warning(
									    "Using deprecated `--per_gpu_train_batch_size` argument which will be removed in a future "
									    "version. Using `--per_device_train_batch_size` is preferred." )
							_snake_case	       =       self.per_gpu_train_batch_size or self.per_device_train_batch_size
							return per_device_batch_size * self.n_replicas






					@property
					def 				__SCREAMING_SNAKE_CASE	(    self ):
							if self.per_gpu_eval_batch_size:
									logger.warning(
									    "Using deprecated `--per_gpu_eval_batch_size` argument which will be removed in a future "
									    "version. Using `--per_device_eval_batch_size` is preferred." )
							_snake_case	       =       self.per_gpu_eval_batch_size or self.per_device_eval_batch_size
							return per_device_batch_size * self.n_replicas






					@property
					def 				__SCREAMING_SNAKE_CASE	(    self ):
							requires_backends(self,  ["tf"] )
							warnings.warn(
							    "The n_gpu argument is deprecated and will be removed in a future version, use n_replicas instead.",  UpperCAmelCase_,  )
							return self._setup_strategy.num_replicas_in_sync







"""simple docstring"""

import torch

from diffusers import KDPMaAncestralDiscreteScheduler
from diffusers.utils import torch_device

from .test_schedulers import SchedulerCommonTest






class        _lowercase   (				__UpperCAmelCase		):
	lowercase_        =	(KDPMaAncestralDiscreteScheduler,)
	lowercase_        =	1_0




	def 			_UpperCamelCase    ( self		, **UpperCAmelCase_					)		->							Optional[int]:
			lowerCamelCase     :			Any		=    {
			    'num_train_timesteps': 1100,
			    'beta_start': 0.0001,
			    'beta_end': 0.02,
			    'beta_schedule': 'linear',
			}

			config.update(**UpperCAmelCase_					)
			return config




	def 			_UpperCamelCase    ( self					)		->							Any:
			for timesteps in [10, 50, 100, 1000]:
					self.check_over_configs(num_train_timesteps=UpperCAmelCase_					)




	def 			_UpperCamelCase    ( self					)		->							List[str]:
			for beta_start, beta_end in zip([0.00001, 0.0001, 0.001]		, [0.0002, 0.002, 0.02]					):
					self.check_over_configs(beta_start=UpperCAmelCase_		, beta_end=UpperCAmelCase_					)




	def 			_UpperCamelCase    ( self					)		->							Optional[int]:
			for schedule in ["linear", "scaled_linear"]:
					self.check_over_configs(beta_schedule=UpperCAmelCase_					)




	def 			_UpperCamelCase    ( self					)		->							List[Any]:
			if torch_device == "mps":
					return
			lowerCamelCase     :			Any		=    self.scheduler_classes[0]
			lowerCamelCase     :			List[str]		=    self.get_scheduler_config()
			lowerCamelCase     :			Union[str, Any]		=    scheduler_class(**UpperCAmelCase_					)

			scheduler.set_timesteps(self.num_inference_steps					)

			lowerCamelCase     :			Optional[Any]		=    torch.manual_seed(0					)

			lowerCamelCase     :			int		=    self.dummy_model()
			lowerCamelCase     :			Tuple		=    self.dummy_sample_deter * scheduler.init_noise_sigma
			lowerCamelCase     :			Union[str, Any]		=    sample.to(UpperCAmelCase_					)

			for i, t in enumerate(scheduler.timesteps					):
					lowerCamelCase     :			Tuple		=    scheduler.scale_model_input(UpperCAmelCase_		, UpperCAmelCase_					)

					lowerCamelCase     :			List[str]		=    model(UpperCAmelCase_		, UpperCAmelCase_					)

					lowerCamelCase     :			Tuple		=    scheduler.step(UpperCAmelCase_		, UpperCAmelCase_		, UpperCAmelCase_		, generator=UpperCAmelCase_					)
					lowerCamelCase     :			str		=    output.prev_sample

			lowerCamelCase     :			Optional[int]		=    torch.sum(torch.abs(UpperCAmelCase_					)					)
			lowerCamelCase     :			Tuple		=    torch.mean(torch.abs(UpperCAmelCase_					)					)

			assert abs(result_sum.item() - 13849.3877					) < 1E-2
			assert abs(result_mean.item() - 18.0331					) < 5E-3




	def 			_UpperCamelCase    ( self					)		->							Optional[int]:
			for prediction_type in ["epsilon", "v_prediction"]:
					self.check_over_configs(prediction_type=UpperCAmelCase_					)




	def 			_UpperCamelCase    ( self					)		->							List[str]:
			if torch_device == "mps":
					return
			lowerCamelCase     :			Dict		=    self.scheduler_classes[0]
			lowerCamelCase     :			Union[str, Any]		=    self.get_scheduler_config(prediction_type='v_prediction'					)
			lowerCamelCase     :			int		=    scheduler_class(**UpperCAmelCase_					)

			scheduler.set_timesteps(self.num_inference_steps					)

			lowerCamelCase     :			int		=    self.dummy_model()
			lowerCamelCase     :			Dict		=    self.dummy_sample_deter * scheduler.init_noise_sigma
			lowerCamelCase     :			Optional[int]		=    sample.to(UpperCAmelCase_					)

			lowerCamelCase     :			Dict		=    torch.manual_seed(0					)

			for i, t in enumerate(scheduler.timesteps					):
					lowerCamelCase     :			Dict		=    scheduler.scale_model_input(UpperCAmelCase_		, UpperCAmelCase_					)

					lowerCamelCase     :			Optional[Any]		=    model(UpperCAmelCase_		, UpperCAmelCase_					)

					lowerCamelCase     :			Tuple		=    scheduler.step(UpperCAmelCase_		, UpperCAmelCase_		, UpperCAmelCase_		, generator=UpperCAmelCase_					)
					lowerCamelCase     :			Optional[Any]		=    output.prev_sample

			lowerCamelCase     :			Union[str, Any]		=    torch.sum(torch.abs(UpperCAmelCase_					)					)
			lowerCamelCase     :			Optional[Any]		=    torch.mean(torch.abs(UpperCAmelCase_					)					)

			assert abs(result_sum.item() - 328.9970					) < 1E-2
			assert abs(result_mean.item() - 0.4284					) < 1E-3




	def 			_UpperCamelCase    ( self					)		->							Any:
			if torch_device == "mps":
					return
			lowerCamelCase     :			Union[str, Any]		=    self.scheduler_classes[0]
			lowerCamelCase     :			Dict		=    self.get_scheduler_config()
			lowerCamelCase     :			Dict		=    scheduler_class(**UpperCAmelCase_					)

			scheduler.set_timesteps(self.num_inference_steps		, device=UpperCAmelCase_					)
			lowerCamelCase     :			int		=    torch.manual_seed(0					)

			lowerCamelCase     :			List[Any]		=    self.dummy_model()
			lowerCamelCase     :			Union[str, Any]		=    self.dummy_sample_deter.to(UpperCAmelCase_					) * scheduler.init_noise_sigma

			for t in scheduler.timesteps:
					lowerCamelCase     :			Optional[int]		=    scheduler.scale_model_input(UpperCAmelCase_		, UpperCAmelCase_					)

					lowerCamelCase     :			str		=    model(UpperCAmelCase_		, UpperCAmelCase_					)

					lowerCamelCase     :			List[str]		=    scheduler.step(UpperCAmelCase_		, UpperCAmelCase_		, UpperCAmelCase_		, generator=UpperCAmelCase_					)
					lowerCamelCase     :			Dict		=    output.prev_sample

			lowerCamelCase     :			List[str]		=    torch.sum(torch.abs(UpperCAmelCase_					)					)
			lowerCamelCase     :			str		=    torch.mean(torch.abs(UpperCAmelCase_					)					)

			assert abs(result_sum.item() - 13849.3818					) < 1E-1
			assert abs(result_mean.item() - 18.0331					) < 1E-3






