
import datasets

from .nmt_bleu import compute_bleu  # From: https://github.com/tensorflow/nmt/blob/master/nmt/scripts/bleu.py


A     :	Dict    = '\\n@INPROCEEDINGS{Papineni02bleu:a,\n    author = {Kishore Papineni and Salim Roukos and Todd Ward and Wei-jing Zhu},\n    title = {BLEU: a Method for Automatic Evaluation of Machine Translation},\n    booktitle = {},\n    year = {2002},\n    pages = {311--318}\n}\n@inproceedings{lin-och-2004-orange,\n    title = "{ORANGE}: a Method for Evaluating Automatic Evaluation Metrics for Machine Translation",\n    author = "Lin, Chin-Yew  and\n      Och, Franz Josef",\n    booktitle = "{COLING} 2004: Proceedings of the 20th International Conference on Computational Linguistics",\n    month = "aug 23{--}aug 27",\n    year = "2004",\n    address = "Geneva, Switzerland",\n    publisher = "COLING",\n    url = "https://www.aclweb.org/anthology/C04-1072",\n    pages = "501--507",\n}\n'

A     :	Union[str, Any]    = '\\nBLEU (bilingual evaluation understudy) is an algorithm for evaluating the quality of text which has been machine-translated from one natural language to another.\nQuality is considered to be the correspondence between a machine\'s output and that of a human: "the closer a machine translation is to a professional human translation,\nthe better it is" – this is the central idea behind BLEU. BLEU was one of the first metrics to claim a high correlation with human judgements of quality, and\nremains one of the most popular automated and inexpensive metrics.\n\nScores are calculated for individual translated segments—generally sentences—by comparing them with a set of good quality reference translations.\nThose scores are then averaged over the whole corpus to reach an estimate of the translation\'s overall quality. Intelligibility or grammatical correctness\nare not taken into account[citation needed].\n\nBLEU\'s output is always a number between 0 and 1. This value indicates how similar the candidate text is to the reference texts, with values closer to 1\nrepresenting more similar texts. Few human translations will attain a score of 1, since this would indicate that the candidate is identical to one of the\nreference translations. For this reason, it is not necessary to attain a score of 1. Because there are more opportunities to match, adding additional\nreference translations will increase the BLEU score.\n'

A     :	Dict    = '\nComputes BLEU score of translated segments against one or more references.\nArgs:\n    predictions: list of translations to score.\n        Each translation should be tokenized into a list of tokens.\n    references: list of lists of references for each translation.\n        Each reference should be tokenized into a list of tokens.\n    max_order: Maximum n-gram order to use when computing BLEU score.\n    smooth: Whether or not to apply Lin et al. 2004 smoothing.\nReturns:\n    \'bleu\': bleu score,\n    \'precisions\': geometric mean of n-gram precisions,\n    \'brevity_penalty\': brevity penalty,\n    \'length_ratio\': ratio of lengths,\n    \'translation_length\': translation_length,\n    \'reference_length\': reference_length\nExamples:\n\n    >>> predictions = [\n    ...     ["hello", "there", "general", "kenobi"],                             # tokenized prediction of the first sample\n    ...     ["foo", "bar", "foobar"]                                             # tokenized prediction of the second sample\n    ... ]\n    >>> references = [\n    ...     [["hello", "there", "general", "kenobi"], ["hello", "there", "!"]],  # tokenized references for the first sample (2 references)\n    ...     [["foo", "bar", "foobar"]]                                           # tokenized references for the second sample (1 reference)\n    ... ]\n    >>> bleu = datasets.load_metric("bleu")\n    >>> results = bleu.compute(predictions=predictions, references=references)\n    >>> print(results["bleu"])\n    1.0\n'







@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION  ,   _KWARGS_DESCRIPTION			)
class 			__A( datasets.Metric			):




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[Any]:


								'''simple docstring'''




								return datasets.MetricInfo(
								    description=_DESCRIPTION		,     citation=_CITATION		,     inputs_description=_KWARGS_DESCRIPTION		,     features=datasets.Features(
								        {
								            '''predictions''': datasets.Sequence(datasets.Value('''string'''		,     id='''token'''				)		,     id='''sequence'''				),
								            '''references''': datasets.Sequence(
								                datasets.Sequence(datasets.Value('''string'''		,     id='''token'''				)		,     id='''sequence'''				)		,     id='''references'''				),
								        }				)		,     codebase_urls=['''https://github.com/tensorflow/nmt/blob/master/nmt/scripts/bleu.py''']		,     reference_urls=[
								        '''https://en.wikipedia.org/wiki/BLEU''',
								        '''https://towardsdatascience.com/evaluating-text-output-in-nlp-bleu-at-your-own-risk-e8609665a213''',
								    ]		,     )




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case		,     _snake_case=4		,     _snake_case=False				)    ->      Any:


								'''simple docstring'''




								__a										=    compute_bleu(
								    reference_corpus=_snake_case		,     translation_corpus=_snake_case		,     max_order=_snake_case		,     smooth=_snake_case				)
								((__a)							,     (__a)							,     (__a)							,     (__a)							,     (__a)							,     (__a))										=    score
								return {
								    "bleu": bleu,
								    "precisions": precisions,
								    "brevity_penalty": bp,
								    "length_ratio": ratio,
								    "translation_length": translation_length,
								    "reference_length": reference_length,
								}

import gc
import unittest

import torch
from parameterized import parameterized

from diffusers import AutoencoderKL
from diffusers.utils import floats_tensor, load_hf_numpy, require_torch_gpu, slow, torch_all_close, torch_device
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import enable_full_determinism

from .test_modeling_common import ModelTesterMixin, UNetTesterMixin


enable_full_determinism()







class 			__A( a  ,   a  ,   unittest.TestCase			):
				snake_case_						   =    AutoencoderKL
				snake_case_						   =    '''sample'''
				snake_case_						   =    1E-2




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[Any]:


								'''simple docstring'''




								__a										=    4
								__a										=    3
								__a										=    (32, 32)

								__a										=    floats_tensor((batch_size, num_channels) + sizes				).to(_snake_case				)

								return {"sample": image}




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[Any]:


								'''simple docstring'''




								return (3, 32, 32)




				@property
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Union[str, Any]:


								'''simple docstring'''




								return (3, 32, 32)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Dict:


								'''simple docstring'''




								__a										=    {
								    '''block_out_channels''': [32, 64],
								    '''in_channels''': 3,
								    '''out_channels''': 3,
								    '''down_block_types''': ['''DownEncoderBlock2D''', '''DownEncoderBlock2D'''],
								    '''up_block_types''': ['''UpDecoderBlock2D''', '''UpDecoderBlock2D'''],
								    '''latent_channels''': 4,
								}
								__a										=    self.dummy_input
								return init_dict, inputs_dict




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Tuple:


								'''simple docstring'''




								pass




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Optional[int]:


								'''simple docstring'''




								pass




				@unittest.skipIf(torch_device == '''mps'''		,     '''Gradient checkpointing skipped on MPS'''				)
				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      int:


								'''simple docstring'''




								__a							,     __a										=    self.prepare_init_args_and_inputs_for_common()
								__a										=    self.model_class(**_snake_case				)
								model.to(_snake_case				)

								assert not model.is_gradient_checkpointing and model.training

								__a										=    model(**_snake_case				).sample
								# run the backwards pass on the model. For backwards pass, for simplicity purpose,
								# we won't calculate the loss and rather backprop on out.sum()
								model.zero_grad()

								__a										=    torch.randn_like(_snake_case				)
								__a										=    (out - labels).mean()
								loss.backward()

								# re-instantiate the model now enabling gradient checkpointing
								__a										=    self.model_class(**_snake_case				)
								# clone model
								model_a.load_state_dict(model.state_dict()				)
								model_a.to(_snake_case				)
								model_a.enable_gradient_checkpointing()

								assert model_a.is_gradient_checkpointing and model_a.training

								__a										=    model_a(**_snake_case				).sample
								# run the backwards pass on the model. For backwards pass, for simplicity purpose,
								# we won't calculate the loss and rather backprop on out.sum()
								model_a.zero_grad()
								__a										=    (out_a - labels).mean()
								loss_a.backward()

								# compare the output and parameters gradients
								self.assertTrue((loss - loss_a).abs() < 1E-5				)
								__a										=    dict(model.named_parameters()				)
								__a										=    dict(model_a.named_parameters()				)
								for name, param in named_params.items():
												self.assertTrue(torch_all_close(param.grad.data		,     named_params_a[name].grad.data		,     atol=5E-5				)				)




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Dict:


								'''simple docstring'''




								__a							,     __a										=    AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''		,     output_loading_info=_snake_case				)
								self.assertIsNotNone(_snake_case				)
								self.assertEqual(len(loading_info['''missing_keys''']				)		,     0				)

								model.to(_snake_case				)
								__a										=    model(**self.dummy_input				)

								assert image is not None, "Make sure output is not None"




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      List[str]:


								'''simple docstring'''




								__a										=    AutoencoderKL.from_pretrained('''fusing/autoencoder-kl-dummy'''				)
								__a										=    model.to(_snake_case				)
								model.eval()

								if torch_device == "mps":
												__a										=    torch.manual_seed(0				)
								else:
												__a										=    torch.Generator(device=_snake_case				).manual_seed(0				)

								__a										=    torch.randn(
								    1		,     model.config.in_channels		,     model.config.sample_size		,     model.config.sample_size		,     generator=torch.manual_seed(0				)		,     )
								__a										=    image.to(_snake_case				)
								with torch.no_grad():
												__a										=    model(_snake_case		,     sample_posterior=_snake_case		,     generator=_snake_case				).sample

								__a										=    output[0, -1, -3:, -3:].flatten().cpu()

								# Since the VAE Gaussian prior's generator is seeded on the appropriate device,
								# the expected output slices are not the same for CPU and GPU.
								if torch_device == "mps":
												__a										=    torch.tensor(
												    [
												        -4.0_078E-01,
												        -3.8_323E-04,
												        -1.2_681E-01,
												        -1.1_462E-01,
												        2.0_095E-01,
												        1.0_893E-01,
												        -8.8_247E-02,
												        -3.0_361E-01,
												        -9.8_644E-03,
												    ]				)
								elif torch_device == "cpu":
												__a										=    torch.tensor(
												    [-0.1352, 0.0878, 0.0419, -0.0818, -0.1069, 0.0688, -0.1458, -0.4446, -0.0026]				)
								else:
												__a										=    torch.tensor(
												    [-0.2421, 0.4642, 0.2507, -0.0438, 0.0682, 0.3160, -0.2018, -0.0727, 0.2485]				)

								self.assertTrue(torch_all_close(_snake_case		,     _snake_case		,     rtol=1E-2				)				)







@slow
class 			__A( unittest.TestCase			):




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case				)    ->      Optional[Any]:


								'''simple docstring'''




								return F"""gaussian_noise_s={seed}_shape={'_'.join([str(_snake_case				) for s in shape]				)}.npy"""




				def 				SCREAMING_SNAKE_CASE_		(     self				)    ->      Dict:


								'''simple docstring'''




								super().tearDown()
								gc.collect()
								torch.cuda.empty_cache()




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case=0		,     _snake_case=(4, 3, 512, 512)		,     _snake_case=False				)    ->      Any:


								'''simple docstring'''




								__a										=    torch.floataa if fpaa else torch.floataa
								__a										=    torch.from_numpy(load_hf_numpy(self.get_file_format(_snake_case		,     _snake_case				)				)				).to(_snake_case				).to(_snake_case				)
								return image




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case="CompVis/stable-diffusion-v1-4"		,     _snake_case=False				)    ->      Optional[Any]:


								'''simple docstring'''




								__a										=    '''fp16''' if fpaa else None
								__a										=    torch.floataa if fpaa else torch.floataa

								__a										=    AutoencoderKL.from_pretrained(
								    _snake_case		,     subfolder='''vae'''		,     torch_dtype=_snake_case		,     revision=_snake_case		,     )
								model.to(_snake_case				).eval()

								return model




				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case=0				)    ->      Tuple:


								'''simple docstring'''




								if torch_device == "mps":
												return torch.manual_seed(_snake_case				)
								return torch.Generator(device=_snake_case				).manual_seed(_snake_case				)




				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.1603, 0.9878, -0.0495, -0.0790, -0.2709, 0.8375, -0.2060, -0.0824], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
				        [47, [-0.2376, 0.1168, 0.1332, -0.4840, -0.2508, -0.0791, -0.0493, -0.4089], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
				        # fmt: on
				    ]				)
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case		,     _snake_case				)    ->      List[Any]:


								'''simple docstring'''




								__a										=    self.get_sd_vae_model()
								__a										=    self.get_sd_image(_snake_case				)
								__a										=    self.get_generator(_snake_case				)

								with torch.no_grad():
												__a										=    model(_snake_case		,     generator=_snake_case		,     sample_posterior=_snake_case				).sample

								assert sample.shape == image.shape

								__a										=    sample[-1, -2:, -2:, :2].flatten().float().cpu()
								__a										=    torch.tensor(expected_slice_mps if torch_device == '''mps''' else expected_slice				)

								assert torch_all_close(_snake_case		,     _snake_case		,     atol=3E-3				)




				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.0513, 0.0289, 1.3799, 0.2166, -0.2573, -0.0871, 0.5103, -0.0999]],
				        [47, [-0.4128, -0.1320, -0.3704, 0.1965, -0.4116, -0.2332, -0.3340, 0.2247]],
				        # fmt: on
				    ]				)
				@require_torch_gpu
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case				)    ->      Tuple:


								'''simple docstring'''




								__a										=    self.get_sd_vae_model(fpaa=_snake_case				)
								__a										=    self.get_sd_image(_snake_case		,     fpaa=_snake_case				)
								__a										=    self.get_generator(_snake_case				)

								with torch.no_grad():
												__a										=    model(_snake_case		,     generator=_snake_case		,     sample_posterior=_snake_case				).sample

								assert sample.shape == image.shape

								__a										=    sample[-1, -2:, :2, -2:].flatten().float().cpu()
								__a										=    torch.tensor(_snake_case				)

								assert torch_all_close(_snake_case		,     _snake_case		,     atol=1E-2				)




				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.1609, 0.9866, -0.0487, -0.0777, -0.2716, 0.8368, -0.2055, -0.0814], [-0.2395, 0.0098, 0.0102, -0.0709, -0.2840, -0.0274, -0.0718, -0.1824]],
				        [47, [-0.2377, 0.1147, 0.1333, -0.4841, -0.2506, -0.0805, -0.0491, -0.4085], [0.0350, 0.0847, 0.0467, 0.0344, -0.0842, -0.0547, -0.0633, -0.1131]],
				        # fmt: on
				    ]				)
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case		,     _snake_case				)    ->      Optional[int]:


								'''simple docstring'''




								__a										=    self.get_sd_vae_model()
								__a										=    self.get_sd_image(_snake_case				)

								with torch.no_grad():
												__a										=    model(_snake_case				).sample

								assert sample.shape == image.shape

								__a										=    sample[-1, -2:, -2:, :2].flatten().float().cpu()
								__a										=    torch.tensor(expected_slice_mps if torch_device == '''mps''' else expected_slice				)

								assert torch_all_close(_snake_case		,     _snake_case		,     atol=3E-3				)




				@parameterized.expand(
				    [
				        # fmt: off
				        [13, [-0.2051, -0.1803, -0.2311, -0.2114, -0.3292, -0.3574, -0.2953, -0.3323]],
				        [37, [-0.2632, -0.2625, -0.2199, -0.2741, -0.4539, -0.4990, -0.3720, -0.4925]],
				        # fmt: on
				    ]				)
				@require_torch_gpu
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case				)    ->      Optional[int]:


								'''simple docstring'''




								__a										=    self.get_sd_vae_model()
								__a										=    self.get_sd_image(_snake_case		,     shape=(3, 4, 64, 64)				)

								with torch.no_grad():
												__a										=    model.decode(_snake_case				).sample

								assert list(sample.shape				) == [3, 3, 512, 512]

								__a										=    sample[-1, -2:, :2, -2:].flatten().cpu()
								__a										=    torch.tensor(_snake_case				)

								assert torch_all_close(_snake_case		,     _snake_case		,     atol=1E-3				)




				@parameterized.expand(
				    [
				        # fmt: off
				        [27, [-0.0369, 0.0207, -0.0776, -0.0682, -0.1747, -0.1930, -0.1465, -0.2039]],
				        [16, [-0.1628, -0.2134, -0.2747, -0.2642, -0.3774, -0.4404, -0.3687, -0.4277]],
				        # fmt: on
				    ]				)
				@require_torch_gpu
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case				)    ->      Optional[Any]:


								'''simple docstring'''




								__a										=    self.get_sd_vae_model(fpaa=_snake_case				)
								__a										=    self.get_sd_image(_snake_case		,     shape=(3, 4, 64, 64)		,     fpaa=_snake_case				)

								with torch.no_grad():
												__a										=    model.decode(_snake_case				).sample

								assert list(sample.shape				) == [3, 3, 512, 512]

								__a										=    sample[-1, -2:, :2, -2:].flatten().float().cpu()
								__a										=    torch.tensor(_snake_case				)

								assert torch_all_close(_snake_case		,     _snake_case		,     atol=5E-3				)




				@parameterized.expand([(13,), (16,), (27,)]				)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()		,     reason='''xformers is not required when using PyTorch 2.0.'''				)
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case				)    ->      Union[str, Any]:


								'''simple docstring'''




								__a										=    self.get_sd_vae_model(fpaa=_snake_case				)
								__a										=    self.get_sd_image(_snake_case		,     shape=(3, 4, 64, 64)		,     fpaa=_snake_case				)

								with torch.no_grad():
												__a										=    model.decode(_snake_case				).sample

								model.enable_xformers_memory_efficient_attention()
								with torch.no_grad():
												__a										=    model.decode(_snake_case				).sample

								assert list(sample.shape				) == [3, 3, 512, 512]

								assert torch_all_close(_snake_case		,     _snake_case		,     atol=1E-1				)




				@parameterized.expand([(13,), (16,), (37,)]				)
				@require_torch_gpu
				@unittest.skipIf(not is_xformers_available()		,     reason='''xformers is not required when using PyTorch 2.0.'''				)
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case				)    ->      List[str]:


								'''simple docstring'''




								__a										=    self.get_sd_vae_model()
								__a										=    self.get_sd_image(_snake_case		,     shape=(3, 4, 64, 64)				)

								with torch.no_grad():
												__a										=    model.decode(_snake_case				).sample

								model.enable_xformers_memory_efficient_attention()
								with torch.no_grad():
												__a										=    model.decode(_snake_case				).sample

								assert list(sample.shape				) == [3, 3, 512, 512]

								assert torch_all_close(_snake_case		,     _snake_case		,     atol=1E-2				)




				@parameterized.expand(
				    [
				        # fmt: off
				        [33, [-0.3001, 0.0918, -2.6984, -3.9720, -3.2099, -5.0353, 1.7338, -0.2065, 3.4267]],
				        [47, [-1.5030, -4.3871, -6.0355, -9.1157, -1.6661, -2.7853, 2.1607, -5.0823, 2.5633]],
				        # fmt: on
				    ]				)
				def 				SCREAMING_SNAKE_CASE_		(     self		,     _snake_case		,     _snake_case				)    ->      Optional[int]:


								'''simple docstring'''




								__a										=    self.get_sd_vae_model()
								__a										=    self.get_sd_image(_snake_case				)
								__a										=    self.get_generator(_snake_case				)

								with torch.no_grad():
												__a										=    model.encode(_snake_case				).latent_dist
												__a										=    dist.sample(generator=_snake_case				)

								assert list(sample.shape				) == [image.shape[0], 4] + [i // 8 for i in image.shape[2:]]

								__a										=    sample[0, -1, -3:, -3:].flatten().cpu()
								__a										=    torch.tensor(_snake_case				)

								__a										=    3E-3 if torch_device != '''mps''' else 1E-2
								assert torch_all_close(_snake_case		,     _snake_case		,     atol=_snake_case				)