
"""simple docstring"""


import re
import tempfile
import unittest
from pathlib import Path

from datasets.utils.metadata import DatasetMetadata






def        lowerCAmelCase_       (		snake_case_		:      str      )       ->str:
			lowerCamelCase__     :					str		=min(re.search('^ +'							,   snake_case_      ).end() if t.startswith(' '      ) else 0 for t in string.splitlines()      )
			return "\n".join([line[indent_level:] for line in string.splitlines() if indent_level < len(snake_case_      )]      )


lowerCAmelCase								=						"""\
---
language:
- zh
- en
task_ids:
- sentiment-classification
---
# Begin of markdown

Some cool dataset card
"""

lowerCAmelCase								=						"""\
---
---
# Begin of markdown

Some cool dataset card
"""


lowerCAmelCase								=						"""\
# Begin of markdown

Some cool dataset card
"""

class    A_     (					unittest.TestCase       ):



							"""simple docstring"""






							def 				UpperCAmelCase__							(  self  :Optional[Any]	):





										"""simple docstring"""

										with tempfile.TemporaryDirectory() as tmp_dir:
													lowerCamelCase__     :					Any		=Path(lowerCamelCase_	) / 'README.md'
													with open(lowerCamelCase_					,							'w+'	) as readme_file:
																readme_file.write(lowerCamelCase_	)
													lowerCamelCase__     :					str		=DatasetMetadata.from_readme(lowerCamelCase_	)
													self.assertDictEqual(lowerCamelCase_					,							{'language': ['zh', 'en'], 'task_ids': ['sentiment-classification']}	)

													with open(lowerCamelCase_					,							'w+'	) as readme_file:
																readme_file.write(lowerCamelCase_	)
													lowerCamelCase__     :					Any		=DatasetMetadata.from_readme(lowerCamelCase_	)
													self.assertDictEqual(lowerCamelCase_					,							{}	)

													with open(lowerCamelCase_					,							'w+'	) as readme_file:
																readme_file.write(lowerCamelCase_	)
													lowerCamelCase__     :					Tuple		=DatasetMetadata.from_readme(lowerCamelCase_	)
													self.assertEqual(lowerCamelCase_					,							{}	)





							def 				UpperCAmelCase__							(  self  :Union[str, Any]	):





										"""simple docstring"""

										lowerCamelCase__     :					Dict		=_dedent(
										    '\\n            annotations_creators:\n            - found\n            language_creators:\n            - found\n            language:\n            - en\n            license:\n            - unknown\n            multilinguality:\n            - monolingual\n            pretty_name: Test Dataset\n            size_categories:\n            - 10K<n<100K\n            source_datasets:\n            - extended|other-yahoo-webscope-l6\n            task_categories:\n            - question-answering\n            task_ids:\n            - open-domain-qa\n            '	)
										assert DatasetMetadata.from_yaml_string(lowerCamelCase_	)

										lowerCamelCase__     :					Union[str, Any]		=_dedent(
										    '\\n            annotations_creators:\n            - found\n            language:\n            - en\n            license:\n            - unknown\n            multilinguality:\n            - monolingual\n            pretty_name: Test Dataset\n            size_categories:\n            - 10K<n<100K\n            source_datasets:\n            - extended|other-yahoo-webscope-l6\n            task_categories:\n            - question-answering\n            task_ids:\n            - open-domain-qa\n            task_ids:\n            - open-domain-qa\n            '	)
										with self.assertRaises(lowerCamelCase_	):
													DatasetMetadata.from_yaml_string(lowerCamelCase_	)

										lowerCamelCase__     :					Optional[int]		=_dedent(
										    '\\n            annotations_creators:\n            - found\n            language_creators:\n            - found\n            language:\n            - en\n            license:\n            - unknown\n            multilinguality:\n            - monolingual\n            pretty_name: Test Dataset\n            size_categories:\n            - 10K<n<100K\n            source_datasets:\n            - extended|other-yahoo-webscope-l6\n            task_categories:\n            - text-classification\n            task_ids:\n            - multi-class-classification\n            paperswithcode_id:\n            - squad\n            configs:\n            - en\n            train-eval-index:\n            - config: en\n              task: text-classification\n              task_id: multi_class_classification\n              splits:\n                train_split: train\n                eval_split: test\n              col_mapping:\n                text: text\n                label: target\n              metrics:\n                - type: accuracy\n                  name: Accuracy\n            extra_gated_prompt: |\n              By clicking on “Access repository” below, you also agree to ImageNet Terms of Access:\n              [RESEARCHER_FULLNAME] (the "Researcher") has requested permission to use the ImageNet database (the "Database") at Princeton University and Stanford University. In exchange for such permission, Researcher hereby agrees to the following terms and conditions:\n              1. Researcher shall use the Database only for non-commercial research and educational purposes.\n            extra_gated_fields:\n              Company: text\n              Country: text\n              I agree to use this model for non-commerical use ONLY: checkbox\n            '	)
										assert DatasetMetadata.from_yaml_string(lowerCamelCase_	)

"""simple docstring"""


import gc
import random
import unittest

import numpy as np
import torch
from PIL import Image
from transformers import CLIPImageProcessor, CLIPVisionConfig, CLIPVisionModelWithProjection

from diffusers import (
    AutoencoderKL,
    DPMSolverMultistepScheduler,
    PNDMScheduler,
    StableDiffusionImageVariationPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, nightly, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import IMAGE_VARIATION_BATCH_PARAMS, IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class    A_     (					A__				, A__				, A__				, unittest.TestCase       ):



							"""simple docstring"""






							SCREAMING_SNAKE_CASE_     				=       StableDiffusionImageVariationPipeline
							SCREAMING_SNAKE_CASE_     				=       IMAGE_VARIATION_PARAMS
							SCREAMING_SNAKE_CASE_     				=       IMAGE_VARIATION_BATCH_PARAMS
							SCREAMING_SNAKE_CASE_     				=       frozenset([]       )
							# TO-DO: update image_params once pipeline is refactored with VaeImageProcessor.preprocess
							SCREAMING_SNAKE_CASE_     				=       frozenset([]       )





							def 				UpperCAmelCase__							(  self  :str	):





										"""simple docstring"""

										torch.manual_seed(0	)
										lowerCamelCase__     :					int		=UNetaDConditionModel(
										    block_out_channels=(32, 64)					,							layers_per_block=2					,							sample_size=32					,							in_channels=4					,							out_channels=4					,							down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')					,							up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')					,							cross_attention_dim=32					,							)
										lowerCamelCase__     :					int		=PNDMScheduler(skip_prk_steps=lowerCamelCase_	)
										torch.manual_seed(0	)
										lowerCamelCase__     :					Optional[int]		=AutoencoderKL(
										    block_out_channels=[32, 64]					,							in_channels=3					,							out_channels=3					,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']					,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']					,							latent_channels=4					,							)
										torch.manual_seed(0	)
										lowerCamelCase__     :					Optional[int]		=CLIPVisionConfig(
										    hidden_size=32					,							projection_dim=32					,							intermediate_size=37					,							layer_norm_eps=1e-05					,							num_attention_heads=4					,							num_hidden_layers=5					,							image_size=32					,							patch_size=4					,							)
										lowerCamelCase__     :					Optional[Any]		=CLIPVisionModelWithProjection(lowerCamelCase_	)
										lowerCamelCase__     :					List[str]		=CLIPImageProcessor(crop_size=32					,							size=32	)

										lowerCamelCase__     :					Optional[int]		={
										    'unet': unet,
										    'scheduler': scheduler,
										    'vae': vae,
										    'image_encoder': image_encoder,
										    'feature_extractor': feature_extractor,
										    'safety_checker': None,
										}
										return components





							def 				UpperCAmelCase__							(  self  :List[Any]					,							lowerCamelCase_  :List[Any]					,							lowerCamelCase_  :Optional[int]=0	):





										"""simple docstring"""

										lowerCamelCase__     :					Optional[Any]		=floats_tensor((1, 3, 32, 32)					,							rng=random.Random(lowerCamelCase_	)	)
										lowerCamelCase__     :					Optional[Any]		=image.cpu().permute(0					,							2					,							3					,							1	)[0]
										lowerCamelCase__     :					List[Any]		=Image.fromarray(np.uinta(lowerCamelCase_	)	).convert('RGB'	).resize((32, 32)	)
										if str(lowerCamelCase_	).startswith('mps'	):
													lowerCamelCase__     :					int		=torch.manual_seed(lowerCamelCase_	)
										else:
													lowerCamelCase__     :					Tuple		=torch.Generator(device=lowerCamelCase_	).manual_seed(lowerCamelCase_	)
										lowerCamelCase__     :					Any		={
										    'image': image,
										    'generator': generator,
										    'num_inference_steps': 2,
										    'guidance_scale': 6.0,
										    'output_type': 'numpy',
										}
										return inputs





							def 				UpperCAmelCase__							(  self  :Optional[int]	):





										"""simple docstring"""

										lowerCamelCase__     :					str		='cpu'  # ensure determinism for the device-dependent torch.Generator
										lowerCamelCase__     :					List[str]		=self.get_dummy_components()
										lowerCamelCase__     :					Union[str, Any]		=StableDiffusionImageVariationPipeline(**lowerCamelCase_	)
										lowerCamelCase__     :					int		=sd_pipe.to(lowerCamelCase_	)
										sd_pipe.set_progress_bar_config(disable=lowerCamelCase_	)

										lowerCamelCase__     :					str		=self.get_dummy_inputs(lowerCamelCase_	)
										lowerCamelCase__     :					List[str]		=sd_pipe(**lowerCamelCase_	).images
										lowerCamelCase__     :					Tuple		=image[0, -3:, -3:, -1]

										assert image.shape == (1, 64, 64, 3)
										lowerCamelCase__     :					str		=np.array([0.52_39, 0.57_23, 0.47_96, 0.50_49, 0.55_50, 0.46_85, 0.53_29, 0.48_91, 0.49_21]	)

										assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-3





							def 				UpperCAmelCase__							(  self  :Tuple	):





										"""simple docstring"""

										lowerCamelCase__     :					Any		='cpu'  # ensure determinism for the device-dependent torch.Generator
										lowerCamelCase__     :					Any		=self.get_dummy_components()
										lowerCamelCase__     :					Any		=StableDiffusionImageVariationPipeline(**lowerCamelCase_	)
										lowerCamelCase__     :					Dict		=sd_pipe.to(lowerCamelCase_	)
										sd_pipe.set_progress_bar_config(disable=lowerCamelCase_	)

										lowerCamelCase__     :					Any		=self.get_dummy_inputs(lowerCamelCase_	)
										lowerCamelCase__     :					Union[str, Any]		=2 * [inputs['image']]
										lowerCamelCase__     :					Optional[int]		=sd_pipe(**lowerCamelCase_	)

										lowerCamelCase__     :					str		=output.images

										lowerCamelCase__     :					str		=image[-1, -3:, -3:, -1]

										assert image.shape == (2, 64, 64, 3)
										lowerCamelCase__     :					int		=np.array([0.68_92, 0.56_37, 0.58_36, 0.57_71, 0.62_54, 0.64_09, 0.55_80, 0.55_69, 0.52_89]	)

										assert np.abs(image_slice.flatten() - expected_slice	).max() < 1e-3





							def 				UpperCAmelCase__							(  self  :List[str]	):





										"""simple docstring"""

										super().test_inference_batch_single_identical(expected_max_diff=3e-3	)






@slow
@require_torch_gpu
class    A_     (					unittest.TestCase       ):



							"""simple docstring"""






							def 				UpperCAmelCase__							(  self  :Union[str, Any]	):





										"""simple docstring"""

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





							def 				UpperCAmelCase__							(  self  :Union[str, Any]					,							lowerCamelCase_  :Optional[Any]					,							lowerCamelCase_  :Optional[Any]="cpu"					,							lowerCamelCase_  :Any=torch.floataa					,							lowerCamelCase_  :str=0	):





										"""simple docstring"""

										lowerCamelCase__     :					Union[str, Any]		=torch.Generator(device=lowerCamelCase_	).manual_seed(lowerCamelCase_	)
										lowerCamelCase__     :					Tuple		=load_image(
										    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
										    '/stable_diffusion_imgvar/input_image_vermeer.png'	)
										lowerCamelCase__     :					str		=np.random.RandomState(lowerCamelCase_	).standard_normal((1, 4, 64, 64)	)
										lowerCamelCase__     :					List[str]		=torch.from_numpy(lowerCamelCase_	).to(device=lowerCamelCase_					,							dtype=lowerCamelCase_	)
										lowerCamelCase__     :					str		={
										    'image': init_image,
										    'latents': latents,
										    'generator': generator,
										    'num_inference_steps': 3,
										    'guidance_scale': 7.5,
										    'output_type': 'numpy',
										}
										return inputs





							def 				UpperCAmelCase__							(  self  :str	):





										"""simple docstring"""

										lowerCamelCase__     :					Tuple		=StableDiffusionImageVariationPipeline.from_pretrained(
										    'lambdalabs/sd-image-variations-diffusers'					,							safety_checker=lowerCamelCase_	)
										lowerCamelCase__     :					List[Any]		=sd_pipe.to(lowerCamelCase_	)
										sd_pipe.set_progress_bar_config(disable=lowerCamelCase_	)

										lowerCamelCase__     :					Union[str, Any]		=self.get_inputs(lowerCamelCase_	)
										lowerCamelCase__     :					Dict		=sd_pipe(**lowerCamelCase_	).images
										lowerCamelCase__     :					Optional[int]		=image[0, -3:, -3:, -1].flatten()

										assert image.shape == (1, 512, 512, 3)
										lowerCamelCase__     :					List[str]		=np.array([0.8_44_91, 0.9_07_89, 0.7_57_08, 0.7_87_34, 0.8_34_85, 0.7_00_99, 0.6_69_38, 0.6_87_27, 0.6_13_79]	)
										assert np.abs(image_slice - expected_slice	).max() < 6e-3





							def 				UpperCAmelCase__							(  self  :Optional[Any]	):





										"""simple docstring"""

										lowerCamelCase__     :					Optional[Any]		=0

										def callback_fn(lowerCamelCase_  :int					,							lowerCamelCase_  :int					,							lowerCamelCase_  :torch.FloatTensor	) -> None:
													lowerCamelCase__     :					Union[str, Any]		=True
													nonlocal number_of_steps
													number_of_steps += 1
													if step == 1:
																lowerCamelCase__     :					Dict		=latents.detach().cpu().numpy()
																assert latents.shape == (1, 4, 64, 64)
																lowerCamelCase__     :					str		=latents[0, -3:, -3:, -1]
																lowerCamelCase__     :					str		=np.array(
																    [-0.16_21, 0.28_37, -0.79_79, -0.12_21, -1.30_57, 0.76_81, -2.11_91, 0.04_64, 1.63_09]	)

																assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5e-2
													elif step == 2:
																lowerCamelCase__     :					Union[str, Any]		=latents.detach().cpu().numpy()
																assert latents.shape == (1, 4, 64, 64)
																lowerCamelCase__     :					Dict		=latents[0, -3:, -3:, -1]
																lowerCamelCase__     :					Optional[Any]		=np.array([0.62_99, 1.75_00, 1.19_92, -2.15_82, -1.89_94, 0.73_34, -0.70_90, 1.01_37, 1.52_73]	)

																assert np.abs(latents_slice.flatten() - expected_slice	).max() < 5e-2

										lowerCamelCase__     :					str		=False

										lowerCamelCase__     :					Optional[Any]		=StableDiffusionImageVariationPipeline.from_pretrained(
										    'fusing/sd-image-variations-diffusers'					,							safety_checker=lowerCamelCase_					,							torch_dtype=torch.floataa					,							)
										pipe.to(lowerCamelCase_	)
										pipe.set_progress_bar_config(disable=lowerCamelCase_	)
										pipe.enable_attention_slicing()

										lowerCamelCase__     :					Any		=self.get_inputs(lowerCamelCase_					,							dtype=torch.floataa	)
										pipe(**lowerCamelCase_					,							callback=lowerCamelCase_					,							callback_steps=1	)
										assert callback_fn.has_been_called
										assert number_of_steps == inputs["num_inference_steps"]





							def 				UpperCAmelCase__							(  self  :List[Any]	):





										"""simple docstring"""

										torch.cuda.empty_cache()
										torch.cuda.reset_max_memory_allocated()
										torch.cuda.reset_peak_memory_stats()

										lowerCamelCase__     :					str		='fusing/sd-image-variations-diffusers'
										lowerCamelCase__     :					Union[str, Any]		=StableDiffusionImageVariationPipeline.from_pretrained(
										    lowerCamelCase_					,							safety_checker=lowerCamelCase_					,							torch_dtype=torch.floataa	)
										lowerCamelCase__     :					Any		=pipe.to(lowerCamelCase_	)
										pipe.set_progress_bar_config(disable=lowerCamelCase_	)
										pipe.enable_attention_slicing(1	)
										pipe.enable_sequential_cpu_offload()

										lowerCamelCase__     :					Dict		=self.get_inputs(lowerCamelCase_					,							dtype=torch.floataa	)
										lowerCamelCase__     :					Tuple		=pipe(**lowerCamelCase_	)

										lowerCamelCase__     :					Any		=torch.cuda.max_memory_allocated()
										# make sure that less than 2.6 GB is allocated
										assert mem_bytes < 2.6 * 10**9






@nightly
@require_torch_gpu
class    A_     (					unittest.TestCase       ):



							"""simple docstring"""






							def 				UpperCAmelCase__							(  self  :Union[str, Any]	):





										"""simple docstring"""

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





							def 				UpperCAmelCase__							(  self  :List[str]					,							lowerCamelCase_  :Optional[int]					,							lowerCamelCase_  :Union[str, Any]="cpu"					,							lowerCamelCase_  :Union[str, Any]=torch.floataa					,							lowerCamelCase_  :Union[str, Any]=0	):





										"""simple docstring"""

										lowerCamelCase__     :					Any		=torch.Generator(device=lowerCamelCase_	).manual_seed(lowerCamelCase_	)
										lowerCamelCase__     :					List[Any]		=load_image(
										    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
										    '/stable_diffusion_imgvar/input_image_vermeer.png'	)
										lowerCamelCase__     :					int		=np.random.RandomState(lowerCamelCase_	).standard_normal((1, 4, 64, 64)	)
										lowerCamelCase__     :					List[str]		=torch.from_numpy(lowerCamelCase_	).to(device=lowerCamelCase_					,							dtype=lowerCamelCase_	)
										lowerCamelCase__     :					str		={
										    'image': init_image,
										    'latents': latents,
										    'generator': generator,
										    'num_inference_steps': 50,
										    'guidance_scale': 7.5,
										    'output_type': 'numpy',
										}
										return inputs





							def 				UpperCAmelCase__							(  self  :Dict	):





										"""simple docstring"""

										lowerCamelCase__     :					List[str]		=StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'	)
										sd_pipe.to(lowerCamelCase_	)
										sd_pipe.set_progress_bar_config(disable=lowerCamelCase_	)

										lowerCamelCase__     :					Optional[Any]		=self.get_inputs(lowerCamelCase_	)
										lowerCamelCase__     :					Dict		=sd_pipe(**lowerCamelCase_	).images[0]

										lowerCamelCase__     :					Optional[int]		=load_numpy(
										    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
										    '/stable_diffusion_imgvar/lambdalabs_variations_pndm.npy'	)
										lowerCamelCase__     :					Dict		=np.abs(expected_image - image	).max()
										assert max_diff < 1e-3





							def 				UpperCAmelCase__							(  self  :Tuple	):





										"""simple docstring"""

										lowerCamelCase__     :					List[Any]		=StableDiffusionImageVariationPipeline.from_pretrained('fusing/sd-image-variations-diffusers'	)
										lowerCamelCase__     :					Optional[Any]		=DPMSolverMultistepScheduler.from_config(sd_pipe.scheduler.config	)
										sd_pipe.to(lowerCamelCase_	)
										sd_pipe.set_progress_bar_config(disable=lowerCamelCase_	)

										lowerCamelCase__     :					Tuple		=self.get_inputs(lowerCamelCase_	)
										lowerCamelCase__     :					int		=25
										lowerCamelCase__     :					Optional[int]		=sd_pipe(**lowerCamelCase_	).images[0]

										lowerCamelCase__     :					str		=load_numpy(
										    'https://huggingface.co/datasets/diffusers/test-arrays/resolve/main'
										    '/stable_diffusion_imgvar/lambdalabs_variations_dpm_multi.npy'	)
										lowerCamelCase__     :					Dict		=np.abs(expected_image - image	).max()
										assert max_diff < 1e-3