





import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
					import torch_xla.core.xla_model as xm
					import torch_xla.debug.metrics as met
class      UpperCamelCase_						(					_lowerCamelCase   ):




					def __init__(						self			,  *lowerCAmelCase_			,  lowerCAmelCase_=None			,  lowerCAmelCase_=None			,  **lowerCAmelCase_					)						->							Tuple:
										super().__init__(*lowerCAmelCase_			,  **lowerCAmelCase_					)
										_snake_case													=			eval_examples
										_snake_case													=			post_process_function




					def 		lowerCAmelCase     (						self			,  lowerCAmelCase_ = None			,  lowerCAmelCase_=None			,  lowerCAmelCase_ = None			,  lowerCAmelCase_ = "eval"			,  **lowerCAmelCase_			,  )						->							Dict[str, float]:
										_snake_case													=			gen_kwargs.copy()
										_snake_case													=			(
										    gen_kwargs['max_length'] if gen_kwargs.get('max_length'					) is not None else self.args.generation_max_length
										)
										_snake_case													=			(
										    gen_kwargs['num_beams'] if gen_kwargs.get('num_beams'					) is not None else self.args.generation_num_beams
										)
										_snake_case													=			gen_kwargs

										_snake_case													=			self.eval_dataset if eval_dataset is None else eval_dataset
										_snake_case													=			self.get_eval_dataloader(lowerCAmelCase_					)
										_snake_case													=			self.eval_examples if eval_examples is None else eval_examples

										# Temporarily disable metric computation, we will do it in the loop here.
										_snake_case													=			self.compute_metrics
										_snake_case													=			None
										_snake_case													=			time.time()
										_snake_case													=			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
										try:
															_snake_case													=			eval_loop(
															    lowerCAmelCase_			,  description='Evaluation'			,  prediction_loss_only=True if compute_metrics is None else None			,  ignore_keys=lowerCAmelCase_			,  metric_key_prefix=lowerCAmelCase_			,  )
										finally:
															_snake_case													=			compute_metrics
										_snake_case													=			self.args.eval_batch_size * self.args.world_size
										if F'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
															start_time += output.metrics[F'''{metric_key_prefix}_jit_compilation_time''']
										output.metrics.update(
										    speed_metrics(
										        lowerCAmelCase_			,  lowerCAmelCase_			,  num_samples=output.num_samples			,  num_steps=math.ceil(output.num_samples / total_batch_size					)			,  )					)

										if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
															# Only the main node write the results by default
															_snake_case													=			self.post_process_function(lowerCAmelCase_			,  lowerCAmelCase_			,  lowerCAmelCase_					)
															_snake_case													=			self.compute_metrics(lowerCAmelCase_					)

															# Prefix all keys with metric_key_prefix + '_'
															for key in list(metrics.keys()					):
																				if not key.startswith(F'''{metric_key_prefix}_'''					):
																									_snake_case													=			metrics.pop(lowerCAmelCase_					)

															metrics.update(output.metrics					)
										else:
															_snake_case													=			output.metrics

										if self.args.should_log:
															# Only the main node log the results by default
															self.log(lowerCAmelCase_					)

										if self.args.tpu_metrics_debug or self.args.debug:
															# tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
															xm.master_print(met.metrics_report()					)

										_snake_case													=			self.callback_handler.on_evaluate(self.args			,  self.state			,  self.control			,  lowerCAmelCase_					)
										return metrics






					def 		lowerCAmelCase     (						self			,  lowerCAmelCase_			,  lowerCAmelCase_			,  lowerCAmelCase_=None			,  lowerCAmelCase_ = "test"			,  **lowerCAmelCase_					)						->							str:
										_snake_case													=			gen_kwargs.copy()

										_snake_case													=			self.get_test_dataloader(lowerCAmelCase_					)

										# Temporarily disable metric computation, we will do it in the loop here.
										_snake_case													=			self.compute_metrics
										_snake_case													=			None
										_snake_case													=			time.time()
										_snake_case													=			self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
										try:
															_snake_case													=			eval_loop(
															    lowerCAmelCase_			,  description='Prediction'			,  prediction_loss_only=True if compute_metrics is None else None			,  ignore_keys=lowerCAmelCase_			,  metric_key_prefix=lowerCAmelCase_			,  )
										finally:
															_snake_case													=			compute_metrics

										_snake_case													=			self.args.eval_batch_size * self.args.world_size
										if F'''{metric_key_prefix}_jit_compilation_time''' in output.metrics:
															start_time += output.metrics[F'''{metric_key_prefix}_jit_compilation_time''']
										output.metrics.update(
										    speed_metrics(
										        lowerCAmelCase_			,  lowerCAmelCase_			,  num_samples=output.num_samples			,  num_steps=math.ceil(output.num_samples / total_batch_size					)			,  )					)
										if self.post_process_function is None or self.compute_metrics is None:
															return output

										_snake_case													=			self.post_process_function(lowerCAmelCase_			,  lowerCAmelCase_			,  lowerCAmelCase_			,  'predict'					)
										_snake_case													=			self.compute_metrics(lowerCAmelCase_					)

										# Prefix all keys with metric_key_prefix + '_'
										for key in list(metrics.keys()					):
															if not key.startswith(F'''{metric_key_prefix}_'''					):
																				_snake_case													=			metrics.pop(lowerCAmelCase_					)
										metrics.update(output.metrics					)
										return PredictionOutput(predictions=predictions.predictions			,  label_ids=predictions.label_ids			,  metrics=lowerCAmelCase_					)










import unittest
from typing import Tuple

import torch

from diffusers.utils import floats_tensor, randn_tensor, torch_all_close, torch_device
from diffusers.utils.testing_utils import require_torch
@require_torch
class      UpperCamelCase_						:




					@property
					def 		lowerCAmelCase     (						self					)						->							int:
										return self.get_dummy_input()




					@property
					def 		lowerCAmelCase     (						self					)						->							Optional[Any]:
										if self.block_type == "down":
															return (4, 32, 16, 16)
										elif self.block_type == "mid":
															return (4, 32, 32, 32)
										elif self.block_type == "up":
															return (4, 32, 64, 64)

										raise ValueError(F'''\'{self.block_type}\' is not a supported block_type. Set it to \'up\', \'mid\', or \'down\'.'''					)




					def 		lowerCAmelCase     (						self			,  lowerCAmelCase_=True			,  lowerCAmelCase_=False			,  lowerCAmelCase_=False			,  lowerCAmelCase_=False			,  )						->							List[str]:
										_snake_case													=			4
										_snake_case													=			32
										_snake_case													=			(32, 32)

										_snake_case													=			torch.manual_seed(0					)
										_snake_case													=			torch.device(lowerCAmelCase_					)
										_snake_case													=			(batch_size, num_channels) + sizes
										_snake_case													=			randn_tensor(lowerCAmelCase_			,  generator=lowerCAmelCase_			,  device=lowerCAmelCase_					)
										_snake_case													=			{'hidden_states': hidden_states}

										if include_temb:
															_snake_case													=			128
															_snake_case													=			randn_tensor((batch_size, temb_channels)			,  generator=lowerCAmelCase_			,  device=lowerCAmelCase_					)

										if include_res_hidden_states_tuple:
															_snake_case													=			torch.manual_seed(1					)
															_snake_case													=			(randn_tensor(lowerCAmelCase_			,  generator=lowerCAmelCase_			,  device=lowerCAmelCase_					),)

										if include_encoder_hidden_states:
															_snake_case													=			floats_tensor((batch_size, 32, 32)					).to(lowerCAmelCase_					)

										if include_skip_sample:
															_snake_case													=			randn_tensor(((batch_size, 3) + sizes)			,  generator=lowerCAmelCase_			,  device=lowerCAmelCase_					)

										return dummy_input




					def 		lowerCAmelCase     (						self					)						->							Tuple:
										_snake_case													=			{
										    'in_channels': 32,
										    'out_channels': 32,
										    'temb_channels': 128,
										}
										if self.block_type == "up":
															_snake_case													=			32

										if self.block_type == "mid":
															init_dict.pop('out_channels'					)

										_snake_case													=			self.dummy_input
										return init_dict, inputs_dict




					def 		lowerCAmelCase     (						self			,  lowerCAmelCase_					)						->							Optional[int]:
										_snake_case							,       _snake_case													=			self.prepare_init_args_and_inputs_for_common()
										_snake_case													=			self.block_class(**lowerCAmelCase_					)
										unet_block.to(lowerCAmelCase_					)
										unet_block.eval()

										with torch.no_grad():
															_snake_case													=			unet_block(**lowerCAmelCase_					)

										if isinstance(lowerCAmelCase_			,  lowerCAmelCase_					):
															_snake_case													=			output[0]

										self.assertEqual(output.shape			,  self.output_shape					)

										_snake_case													=			output[0, -1, -3:, -3:]
										_snake_case													=			torch.tensor(lowerCAmelCase_					).to(lowerCAmelCase_					)
										assert torch_all_close(output_slice.flatten()			,  lowerCAmelCase_			,  atol=5E-3					)






					@unittest.skipIf(torch_device == 'mps'			,  'Training is not supported in mps'					)
					def 		lowerCAmelCase     (						self					)						->							Tuple:
										_snake_case							,       _snake_case													=			self.prepare_init_args_and_inputs_for_common()
										_snake_case													=			self.block_class(**lowerCAmelCase_					)
										model.to(lowerCAmelCase_					)
										model.train()
										_snake_case													=			model(**lowerCAmelCase_					)

										if isinstance(lowerCAmelCase_			,  lowerCAmelCase_					):
															_snake_case													=			output[0]

										_snake_case													=			torch.device(lowerCAmelCase_					)
										_snake_case													=			randn_tensor(output.shape			,  device=lowerCAmelCase_					)
										_snake_case													=			torch.nn.functional.mse_loss(lowerCAmelCase_			,  lowerCAmelCase_					)
										loss.backward()



