
from __future__ import annotations

import unittest

from transformers import is_tf_available
from transformers.testing_utils import require_tf, slow

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
							import numpy
							import tensorflow as tf

							from transformers import (
							    TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
							    TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST,
							    TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST,
							    BertConfig,
							    DPRConfig,
							    TFDPRContextEncoder,
							    TFDPRQuestionEncoder,
							    TFDPRReader,
							)







class        lowercase	:




							def __init__(       self     ,A__     ,A__=1_3     ,A__=7     ,A__=True     ,A__=True     ,A__=True     ,A__=True     ,A__=9_9     ,A__=3_2     ,A__=2     ,A__=4     ,A__=3_7     ,A__="gelu"     ,A__=0.1     ,A__=0.1     ,A__=5_1_2     ,A__=1_6     ,A__=2     ,A__=0.02     ,A__=3     ,A__=4     ,A__=None     ,A__=0     ,):
												lowercase					=		parent
												lowercase					=		batch_size
												lowercase					=		seq_length
												lowercase					=		is_training
												lowercase					=		use_input_mask
												lowercase					=		use_token_type_ids
												lowercase					=		use_labels
												lowercase					=		vocab_size
												lowercase					=		hidden_size
												lowercase					=		num_hidden_layers
												lowercase					=		num_attention_heads
												lowercase					=		intermediate_size
												lowercase					=		hidden_act
												lowercase					=		hidden_dropout_prob
												lowercase					=		attention_probs_dropout_prob
												lowercase					=		max_position_embeddings
												lowercase					=		type_vocab_size
												lowercase					=		type_sequence_label_size
												lowercase					=		initializer_range
												lowercase					=		num_labels
												lowercase					=		num_choices
												lowercase					=		scope
												lowercase					=		projection_dim




							def 				A__   (       self):
												lowercase					=		ids_tensor([self.batch_size, self.seq_length]     ,self.vocab_size)

												lowercase					=		None
												if self.use_input_mask:
																	# follow test_modeling_tf_ctrl.py
																	lowercase					=		random_attention_mask([self.batch_size, self.seq_length])

												lowercase					=		None
												if self.use_token_type_ids:
																	lowercase					=		ids_tensor([self.batch_size, self.seq_length]     ,self.type_vocab_size)

												lowercase					=		None
												lowercase					=		None
												lowercase					=		None
												if self.use_labels:
																	lowercase					=		ids_tensor([self.batch_size]     ,self.type_sequence_label_size)
																	lowercase					=		ids_tensor([self.batch_size, self.seq_length]     ,self.num_labels)
																	lowercase					=		ids_tensor([self.batch_size]     ,self.num_choices)

												lowercase					=		BertConfig(
												    vocab_size=self.vocab_size     ,hidden_size=self.hidden_size     ,num_hidden_layers=self.num_hidden_layers     ,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     ,max_position_embeddings=self.max_position_embeddings     ,type_vocab_size=self.type_vocab_size     ,is_decoder=A__     ,initializer_range=self.initializer_range     ,)
												lowercase					=		DPRConfig(projection_dim=self.projection_dim     ,**config.to_dict())

												return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels




							def 				A__   (       self     ,A__     ,A__     ,A__     ,A__     ,A__     ,A__     ,A__):
												lowercase					=		TFDPRContextEncoder(config=A__)
												lowercase					=		model(A__     ,attention_mask=A__     ,token_type_ids=A__)
												lowercase					=		model(A__     ,token_type_ids=A__)
												lowercase					=		model(A__)
												self.parent.assertEqual(result.pooler_output.shape     ,(self.batch_size, self.projection_dim or self.hidden_size))




							def 				A__   (       self     ,A__     ,A__     ,A__     ,A__     ,A__     ,A__     ,A__):
												lowercase					=		TFDPRQuestionEncoder(config=A__)
												lowercase					=		model(A__     ,attention_mask=A__     ,token_type_ids=A__)
												lowercase					=		model(A__     ,token_type_ids=A__)
												lowercase					=		model(A__)
												self.parent.assertEqual(result.pooler_output.shape     ,(self.batch_size, self.projection_dim or self.hidden_size))




							def 				A__   (       self     ,A__     ,A__     ,A__     ,A__     ,A__     ,A__     ,A__):
												lowercase					=		TFDPRReader(config=A__)
												lowercase					=		model(A__     ,attention_mask=A__)

												self.parent.assertEqual(result.start_logits.shape     ,(self.batch_size, self.seq_length))
												self.parent.assertEqual(result.end_logits.shape     ,(self.batch_size, self.seq_length))
												self.parent.assertEqual(result.relevance_logits.shape     ,(self.batch_size,))





							def 				A__   (       self):
												lowercase					=		self.prepare_config_and_inputs()
												(
												    (
												    lowercase
												)			,		(
												    lowercase
												)			,		(
												    lowercase
												)			,		(
												    lowercase
												)			,		(
												    lowercase
												)			,		(
												    lowercase
												)			,		(
												    lowercase
												)			,		
												)					=		config_and_inputs
												lowercase					=		{'''input_ids''': input_ids}
												return config, inputs_dict







@require_tf
class        lowercase	(      SCREAMING_SNAKE_CASE__ ,						SCREAMING_SNAKE_CASE__ ,						unittest.TestCase	):
							lowercase_ : Optional[Any]       =(
							    (
							        TFDPRContextEncoder,
							        TFDPRQuestionEncoder,
							        TFDPRReader,
							    )
							    if is_tf_available()
							    else ()
							)
							lowercase_ : List[Any]       ={'''feature-extraction''': TFDPRQuestionEncoder} if is_tf_available() else {}

							lowercase_ : List[Any]       =False
							lowercase_ : int       =False
							lowercase_ : Union[str, Any]       =False
							lowercase_ : int       =False
							lowercase_ : List[Any]       =False




							def 				A__   (       self):
												lowercase					=		TFDPRModelTester(self)
												lowercase					=		ConfigTester(self     ,config_class=A__     ,hidden_size=3_7)




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




							def 				A__   (       self):
												lowercase					=		self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_dpr_context_encoder(*A__)




							def 				A__   (       self):
												lowercase					=		self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_dpr_question_encoder(*A__)




							def 				A__   (       self):
												lowercase					=		self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_dpr_reader(*A__)





							@slow
							def 				A__   (       self):
												for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	lowercase					=		TFDPRContextEncoder.from_pretrained(A__)
																	self.assertIsNotNone(A__)

												for model_name in TF_DPR_CONTEXT_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	lowercase					=		TFDPRContextEncoder.from_pretrained(A__)
																	self.assertIsNotNone(A__)

												for model_name in TF_DPR_QUESTION_ENCODER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	lowercase					=		TFDPRQuestionEncoder.from_pretrained(A__)
																	self.assertIsNotNone(A__)

												for model_name in TF_DPR_READER_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
																	lowercase					=		TFDPRReader.from_pretrained(A__)
																	self.assertIsNotNone(A__)







@require_tf
class        lowercase	(      unittest.TestCase	):




							@slow
							def 				A__   (       self):
												lowercase					=		TFDPRQuestionEncoder.from_pretrained('''facebook/dpr-question_encoder-single-nq-base''')

												lowercase					=		tf.constant(
												    [[1_0_1, 7_5_9_2, 1_0_1_0, 2_0_0_3, 2_0_2_6, 3_8_9_9, 1_0_1_4_0, 1_0_2_9, 1_0_2]])  # [CLS] hello, is my dog cute? [SEP]
												lowercase					=		model(A__)[0]  # embedding shape = (1, 768)
												# compare the actual values for a slice.
												lowercase					=		tf.constant(
												    [
												        [
												            0.03236253,
												            0.12753335,
												            0.16818509,
												            0.00279786,
												            0.3896933,
												            0.24264945,
												            0.2178971,
												            -0.02335227,
												            -0.08481959,
												            -0.14324117,
												        ]
												    ])
												self.assertTrue(numpy.allclose(output[:, :1_0].numpy()     ,expected_slice.numpy()     ,atol=1E-4))






import tempfile

import torch

from diffusers import (
    DEISMultistepScheduler,
    DPMSolverMultistepScheduler,
    DPMSolverSinglestepScheduler,
    UniPCMultistepScheduler,
)

from .test_schedulers import SchedulerCommonTest







class        lowercase	(      SCREAMING_SNAKE_CASE__	):
							lowercase_ : List[str]       =(UniPCMultistepScheduler,)
							lowercase_ : Tuple       =(('''num_inference_steps''', 25),)




							def 				A__   (       self     ,**A__):
												lowercase					=		{
												    '''num_train_timesteps''': 1_0_0_0,
												    '''beta_start''': 0.0001,
												    '''beta_end''': 0.02,
												    '''beta_schedule''': '''linear''',
												    '''solver_order''': 2,
												    '''solver_type''': '''bh2''',
												}

												config.update(**A__)
												return config




							def 				A__   (       self     ,A__=0     ,**A__):
												lowercase					=		dict(self.forward_default_kwargs)
												lowercase					=		kwargs.pop('''num_inference_steps'''     ,A__)
												lowercase					=		self.dummy_sample
												lowercase					=		0.1 * sample
												lowercase					=		[residual + 0.2, residual + 0.15, residual + 0.10]

												for scheduler_class in self.scheduler_classes:
																	lowercase					=		self.get_scheduler_config(**A__)
																	lowercase					=		scheduler_class(**A__)
																	scheduler.set_timesteps(A__)
																	# copy over dummy past residuals
																	lowercase					=		dummy_past_residuals[: scheduler.config.solver_order]

																	with tempfile.TemporaryDirectory() as tmpdirname:
																						scheduler.save_config(A__)
																						lowercase					=		scheduler_class.from_pretrained(A__)
																						new_scheduler.set_timesteps(A__)
																						# copy over dummy past residuals
																						lowercase					=		dummy_past_residuals[: new_scheduler.config.solver_order]

																	lowercase			,		lowercase					=		sample, sample
																	for t in range(A__     ,time_step + scheduler.config.solver_order + 1):
																						lowercase					=		scheduler.step(A__     ,A__     ,A__     ,**A__).prev_sample
																						lowercase					=		new_scheduler.step(A__     ,A__     ,A__     ,**A__).prev_sample

																						assert torch.sum(torch.abs(output - new_output)) < 1E-5, "Scheduler outputs are not identical"




							def 				A__   (       self     ,A__=0     ,**A__):
												lowercase					=		dict(self.forward_default_kwargs)
												lowercase					=		kwargs.pop('''num_inference_steps'''     ,A__)
												lowercase					=		self.dummy_sample
												lowercase					=		0.1 * sample
												lowercase					=		[residual + 0.2, residual + 0.15, residual + 0.10]

												for scheduler_class in self.scheduler_classes:
																	lowercase					=		self.get_scheduler_config()
																	lowercase					=		scheduler_class(**A__)
																	scheduler.set_timesteps(A__)

																	# copy over dummy past residuals (must be after setting timesteps)
																	lowercase					=		dummy_past_residuals[: scheduler.config.solver_order]

																	with tempfile.TemporaryDirectory() as tmpdirname:
																						scheduler.save_config(A__)
																						lowercase					=		scheduler_class.from_pretrained(A__)
																						# copy over dummy past residuals
																						new_scheduler.set_timesteps(A__)

																						# copy over dummy past residual (must be after setting timesteps)
																						lowercase					=		dummy_past_residuals[: new_scheduler.config.solver_order]

																	lowercase					=		scheduler.step(A__     ,A__     ,A__     ,**A__).prev_sample
																	lowercase					=		new_scheduler.step(A__     ,A__     ,A__     ,**A__).prev_sample

																	assert torch.sum(torch.abs(output - new_output)) < 1E-5, "Scheduler outputs are not identical"




							def 				A__   (       self     ,A__=None     ,**A__):
												if scheduler is None:
																	lowercase					=		self.scheduler_classes[0]
																	lowercase					=		self.get_scheduler_config(**A__)
																	lowercase					=		scheduler_class(**A__)

												lowercase					=		self.scheduler_classes[0]
												lowercase					=		self.get_scheduler_config(**A__)
												lowercase					=		scheduler_class(**A__)

												lowercase					=		1_0
												lowercase					=		self.dummy_model()
												lowercase					=		self.dummy_sample_deter
												scheduler.set_timesteps(A__)

												for i, t in enumerate(scheduler.timesteps):
																	lowercase					=		model(A__     ,A__)
																	lowercase					=		scheduler.step(A__     ,A__     ,A__).prev_sample

												return sample




							def 				A__   (       self):
												lowercase					=		dict(self.forward_default_kwargs)

												lowercase					=		kwargs.pop('''num_inference_steps'''     ,A__)

												for scheduler_class in self.scheduler_classes:
																	lowercase					=		self.get_scheduler_config()
																	lowercase					=		scheduler_class(**A__)

																	lowercase					=		self.dummy_sample
																	lowercase					=		0.1 * sample

																	if num_inference_steps is not None and hasattr(A__     ,'''set_timesteps'''):
																						scheduler.set_timesteps(A__)
																	elif num_inference_steps is not None and not hasattr(A__     ,'''set_timesteps'''):
																						lowercase					=		num_inference_steps

																	# copy over dummy past residuals (must be done after set_timesteps)
																	lowercase					=		[residual + 0.2, residual + 0.15, residual + 0.10]
																	lowercase					=		dummy_past_residuals[: scheduler.config.solver_order]

																	lowercase					=		scheduler.timesteps[5]
																	lowercase					=		scheduler.timesteps[6]

																	lowercase					=		scheduler.step(A__     ,A__     ,A__     ,**A__).prev_sample
																	lowercase					=		scheduler.step(A__     ,A__     ,A__     ,**A__).prev_sample

																	self.assertEqual(output_a.shape     ,sample.shape)
																	self.assertEqual(output_a.shape     ,output_a.shape)




							def 				A__   (       self):
												# make sure that iterating over schedulers with same config names gives same results
												# for defaults
												lowercase					=		UniPCMultistepScheduler(**self.get_scheduler_config())
												lowercase					=		self.full_loop(scheduler=A__)
												lowercase					=		torch.mean(torch.abs(A__))

												assert abs(result_mean.item() - 0.2464) < 1E-3

												lowercase					=		DPMSolverSinglestepScheduler.from_config(scheduler.config)
												lowercase					=		DEISMultistepScheduler.from_config(scheduler.config)
												lowercase					=		DPMSolverMultistepScheduler.from_config(scheduler.config)
												lowercase					=		UniPCMultistepScheduler.from_config(scheduler.config)

												lowercase					=		self.full_loop(scheduler=A__)
												lowercase					=		torch.mean(torch.abs(A__))

												assert abs(result_mean.item() - 0.2464) < 1E-3




							def 				A__   (       self):
												for timesteps in [2_5, 5_0, 1_0_0, 9_9_9, 1_0_0_0]:
																	self.check_over_configs(num_train_timesteps=A__)




							def 				A__   (       self):
												self.check_over_configs(thresholding=A__)
												for order in [1, 2, 3]:
																	for solver_type in ["bh1", "bh2"]:
																						for threshold in [0.5, 1.0, 2.0]:
																											for prediction_type in ["epsilon", "sample"]:
																																self.check_over_configs(
																																    thresholding=A__     ,prediction_type=A__     ,sample_max_value=A__     ,solver_order=A__     ,solver_type=A__     ,)




							def 				A__   (       self):
												for prediction_type in ["epsilon", "v_prediction"]:
																	self.check_over_configs(prediction_type=A__)




							def 				A__   (       self):
												for solver_type in ["bh1", "bh2"]:
																	for order in [1, 2, 3]:
																						for prediction_type in ["epsilon", "sample"]:
																											self.check_over_configs(
																											    solver_order=A__     ,solver_type=A__     ,prediction_type=A__     ,)
																											lowercase					=		self.full_loop(
																											    solver_order=A__     ,solver_type=A__     ,prediction_type=A__     ,)
																											assert not torch.isnan(A__).any(), "Samples have nan numbers"




							def 				A__   (       self):
												self.check_over_configs(lower_order_final=A__)
												self.check_over_configs(lower_order_final=A__)




							def 				A__   (       self):
												for num_inference_steps in [1, 2, 3, 5, 1_0, 5_0, 1_0_0, 9_9_9, 1_0_0_0]:
																	self.check_over_forward(num_inference_steps=A__     ,time_step=0)




							def 				A__   (       self):
												lowercase					=		self.full_loop()
												lowercase					=		torch.mean(torch.abs(A__))

												assert abs(result_mean.item() - 0.2464) < 1E-3




							def 				A__   (       self):
												lowercase					=		self.full_loop(prediction_type='''v_prediction''')
												lowercase					=		torch.mean(torch.abs(A__))

												assert abs(result_mean.item() - 0.1014) < 1E-3




							def 				A__   (       self):
												lowercase					=		self.scheduler_classes[0]
												lowercase					=		self.get_scheduler_config(thresholding=A__     ,dynamic_thresholding_ratio=0)
												lowercase					=		scheduler_class(**A__)

												lowercase					=		1_0
												lowercase					=		self.dummy_model()
												lowercase					=		self.dummy_sample_deter.half()
												scheduler.set_timesteps(A__)

												for i, t in enumerate(scheduler.timesteps):
																	lowercase					=		model(A__     ,A__)
																	lowercase					=		scheduler.step(A__     ,A__     ,A__).prev_sample

												assert sample.dtype == torch.floataa





							def 				A__   (       self     ,**A__):
												for scheduler_class in self.scheduler_classes:
																	lowercase					=		self.get_scheduler_config(**A__)
																	lowercase					=		scheduler_class(**A__)

																	scheduler.set_timesteps(scheduler.config.num_train_timesteps)
																	assert len(scheduler.timesteps.unique()) == scheduler.num_inference_steps




