

import gc
import random
import unittest

import numpy as np
import torch
from transformers import XLMRobertaTokenizer

from diffusers import (
    AltDiffusionImgaImgPipeline,
    AutoencoderKL,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.image_processor import VaeImageProcessor
from diffusers.pipelines.alt_diffusion.modeling_roberta_series import (
    RobertaSeriesConfig,
    RobertaSeriesModelWithTransformation,
)
from diffusers.utils import floats_tensor, load_image, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu


enable_full_determinism()

class 							UpperCAmelCase_						(     unittest.TestCase  ):
							"""simple docstring"""







							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												# clean up the VRAM after each test
												super().tearDown()
												gc.collect()
												torch.cuda.empty_cache()


							@property
							def 							__lowercase							(     self			)       ->       Dict:
												_a      :       Optional[Any]   =						1
												_a      :       str   =						3
												_a      :       List[str]   =						(3_2, 3_2)

												_a      :       Tuple   =						floats_tensor((batch_size, num_channels) + sizes			,				rng=random.Random(0			)			).to(_a			)
												return image


							@property
							def 							__lowercase							(     self			)       ->       str:
												torch.manual_seed(0			)
												_a      :       Dict   =						UNetaDConditionModel(
												    block_out_channels=(3_2, 6_4)			,				layers_per_block=2			,				sample_size=3_2			,				in_channels=4			,				out_channels=4			,				down_block_types=('''DownBlock2D''', '''CrossAttnDownBlock2D''')			,				up_block_types=('''CrossAttnUpBlock2D''', '''UpBlock2D''')			,				cross_attention_dim=3_2			,				)
												return model


							@property
							def 							__lowercase							(     self			)       ->       List[str]:
												torch.manual_seed(0			)
												_a      :       Any   =						AutoencoderKL(
												    block_out_channels=[3_2, 6_4]			,				in_channels=3			,				out_channels=3			,				down_block_types=['''DownEncoderBlock2D''', '''DownEncoderBlock2D''']			,				up_block_types=['''UpDecoderBlock2D''', '''UpDecoderBlock2D''']			,				latent_channels=4			,				)
												return model


							@property
							def 							__lowercase							(     self			)       ->       int:
												torch.manual_seed(0			)
												_a      :       Optional[Any]   =						RobertaSeriesConfig(
												    hidden_size=3_2			,				project_dim=3_2			,				intermediate_size=3_7			,				layer_norm_eps=1e-0_5			,				num_attention_heads=4			,				num_hidden_layers=5			,				pad_token_id=1			,				vocab_size=5_0_0_6			,				)
												return RobertaSeriesModelWithTransformation(_a			)


							@property
							def 							__lowercase							(     self			)       ->       List[Any]:
												def extract(*_a			,				**_a			):

																	class 							UpperCAmelCase_						:
																								"""simple docstring"""







																								def __init__(     self			)       ->       Union[str, Any]:
																													_a      :       Union[str, Any]   =						torch.ones([0]			)




																								def 							__lowercase							(     self			,				_a			)       ->       Union[str, Any]:
																													self.pixel_values.to(_a			)
																													return self



																	return Out()

												return extract


							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												_a      :       Union[str, Any]   =						'''cpu'''  # ensure determinism for the device-dependent torch.Generator
												_a      :       int   =						self.dummy_cond_unet
												_a      :       List[Any]   =						PNDMScheduler(skip_prk_steps=_a			)
												_a      :       Optional[Any]   =						self.dummy_vae
												_a      :       List[str]   =						self.dummy_text_encoder
												_a      :       List[Any]   =						XLMRobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-xlm-roberta'''			)
												_a      :       int   =						7_7

												_a      :       List[str]   =						self.dummy_image.to(_a			)
												_a      :       str   =						init_image / 2 + 0.5

												# make sure here that pndm scheduler skips prk
												_a      :       int   =						AltDiffusionImgaImgPipeline(
												    unet=_a			,				scheduler=_a			,				vae=_a			,				text_encoder=_a			,				tokenizer=_a			,				safety_checker=_a			,				feature_extractor=self.dummy_extractor			,				)
												_a      :       Optional[int]   =						VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor			,				do_normalize=_a			)
												_a      :       str   =						alt_pipe.to(_a			)
												alt_pipe.set_progress_bar_config(disable=_a			)

												_a      :       Optional[int]   =						'''A painting of a squirrel eating a burger'''
												_a      :       Dict   =						torch.Generator(device=_a			).manual_seed(0			)
												_a      :       Optional[Any]   =						alt_pipe(
												    [prompt]			,				generator=_a			,				guidance_scale=6.0			,				num_inference_steps=2			,				output_type='''np'''			,				image=_a			,				)

												_a      :       int   =						output.images

												_a      :       List[Any]   =						torch.Generator(device=_a			).manual_seed(0			)
												_a      :       Union[str, Any]   =						alt_pipe(
												    [prompt]			,				generator=_a			,				guidance_scale=6.0			,				num_inference_steps=2			,				output_type='''np'''			,				image=_a			,				return_dict=_a			,				)[0]

												_a      :       int   =						image[0, -3:, -3:, -1]
												_a      :       List[str]   =						image_from_tuple[0, -3:, -3:, -1]

												assert image.shape == (1, 3_2, 3_2, 3)
												_a      :       Optional[int]   =						np.array([0.4427, 0.3731, 0.4249, 0.4941, 0.4546, 0.4148, 0.4193, 0.4666, 0.4499]			)

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


							@unittest.skipIf(torch_device != '''cuda'''			,				'''This test requires a GPU'''			)
							def 							__lowercase							(     self			)       ->       List[str]:
												_a      :       int   =						self.dummy_cond_unet
												_a      :       List[Any]   =						PNDMScheduler(skip_prk_steps=_a			)
												_a      :       List[str]   =						self.dummy_vae
												_a      :       str   =						self.dummy_text_encoder
												_a      :       List[str]   =						XLMRobertaTokenizer.from_pretrained('''hf-internal-testing/tiny-xlm-roberta'''			)
												_a      :       int   =						7_7

												_a      :       Optional[int]   =						self.dummy_image.to(_a			)

												# put models in fp16
												_a      :       int   =						unet.half()
												_a      :       int   =						vae.half()
												_a      :       List[str]   =						bert.half()

												# make sure here that pndm scheduler skips prk
												_a      :       List[str]   =						AltDiffusionImgaImgPipeline(
												    unet=_a			,				scheduler=_a			,				vae=_a			,				text_encoder=_a			,				tokenizer=_a			,				safety_checker=_a			,				feature_extractor=self.dummy_extractor			,				)
												_a      :       Any   =						VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor			,				do_normalize=_a			)
												_a      :       Any   =						alt_pipe.to(_a			)
												alt_pipe.set_progress_bar_config(disable=_a			)

												_a      :       List[str]   =						'''A painting of a squirrel eating a burger'''
												_a      :       Optional[Any]   =						torch.manual_seed(0			)
												_a      :       List[str]   =						alt_pipe(
												    [prompt]			,				generator=_a			,				num_inference_steps=2			,				output_type='''np'''			,				image=_a			,				).images

												assert image.shape == (1, 3_2, 3_2, 3)




							@unittest.skipIf(torch_device != '''cuda'''			,				'''This test requires a GPU'''			)
							def 							__lowercase							(     self			)       ->       str:
												_a      :       Tuple   =						load_image(
												    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
												    '''/img2img/sketch-mountains-input.jpg'''			)
												# resize to resolution that is divisible by 8 but not 16 or 32
												_a      :       str   =						init_image.resize((7_6_0, 5_0_4)			)

												_a      :       Union[str, Any]   =						'''BAAI/AltDiffusion'''
												_a      :       Any   =						AltDiffusionImgaImgPipeline.from_pretrained(
												    _a			,				safety_checker=_a			,				)
												pipe.to(_a			)
												pipe.set_progress_bar_config(disable=_a			)
												pipe.enable_attention_slicing()

												_a      :       Any   =						'''A fantasy landscape, trending on artstation'''

												_a      :       str   =						torch.manual_seed(0			)
												_a      :       Any   =						pipe(
												    prompt=_a			,				image=_a			,				strength=0.75			,				guidance_scale=7.5			,				generator=_a			,				output_type='''np'''			,				)
												_a      :       List[Any]   =						output.images[0]

												_a      :       List[str]   =						image[2_5_5:2_5_8, 3_8_3:3_8_6, -1]

												assert image.shape == (5_0_4, 7_6_0, 3)
												_a      :       Optional[Any]   =						np.array([0.9358, 0.9397, 0.9599, 0.9901, 1.0000, 1.0000, 0.9882, 1.0000, 1.0000]			)

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



@slow
@require_torch_gpu
class 							UpperCAmelCase_						(     unittest.TestCase  ):
							"""simple docstring"""







							def 							__lowercase							(     self			)       ->       Optional[Any]:
												# clean up the VRAM after each test
												super().tearDown()
												gc.collect()
												torch.cuda.empty_cache()




							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												_a      :       Any   =						load_image(
												    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main'''
												    '''/img2img/sketch-mountains-input.jpg'''			)
												_a      :       str   =						init_image.resize((7_6_8, 5_1_2)			)
												_a      :       str   =						load_numpy(
												    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/img2img/fantasy_landscape_alt.npy'''			)

												_a      :       Dict   =						'''BAAI/AltDiffusion'''
												_a      :       List[Any]   =						AltDiffusionImgaImgPipeline.from_pretrained(
												    _a			,				safety_checker=_a			,				)
												pipe.to(_a			)
												pipe.set_progress_bar_config(disable=_a			)
												pipe.enable_attention_slicing()

												_a      :       int   =						'''A fantasy landscape, trending on artstation'''

												_a      :       str   =						torch.manual_seed(0			)
												_a      :       Optional[Any]   =						pipe(
												    prompt=_a			,				image=_a			,				strength=0.75			,				guidance_scale=7.5			,				generator=_a			,				output_type='''np'''			,				)
												_a      :       Any   =						output.images[0]

												assert image.shape == (5_1_2, 7_6_8, 3)
												# img2img is flaky across GPUs even in fp32, so using MAE here
												assert np.abs(expected_image - image			).max() < 1e-2






import json
import sys
import tempfile
import unittest
from pathlib import Path

import transformers
from transformers import (
    CONFIG_MAPPING,
    IMAGE_PROCESSOR_MAPPING,
    AutoConfig,
    AutoImageProcessor,
    CLIPConfig,
    CLIPImageProcessor,
)
from transformers.testing_utils import DUMMY_UNKNOWN_IDENTIFIER


sys.path.append(str(Path(__file__).parent.parent.parent.parent / '''utils'''))

from test_module.custom_configuration import CustomConfig  # noqa E402
from test_module.custom_image_processing import CustomImageProcessor  # noqa E402

class 							UpperCAmelCase_						(     unittest.TestCase  ):
							"""simple docstring"""







							def 							__lowercase							(     self			)       ->       List[Any]:
												_a      :       int   =						0


							def 							__lowercase							(     self			)       ->       List[str]:
												_a      :       Dict   =						AutoImageProcessor.from_pretrained('''openai/clip-vit-base-patch32'''			)
												self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Tuple   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       Optional[Any]   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Optional[Any]:
												# Ensure we can load the image processor from the feature extractor config
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Optional[int]   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       Any   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''feature_extractor_type''': '''CLIPFeatureExtractor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Optional[Any]:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Dict   =						CLIPConfig()

																	# Create a dummy config file with image_proceesor_type
																	_a      :       Tuple   =						Path(_a			) / '''preprocessor_config.json'''
																	_a      :       List[str]   =						Path(_a			) / '''config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																	json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																	# remove image_processor_type to make sure config.json alone is enough to load image processor locally
																	_a      :       Tuple   =						AutoImageProcessor.from_pretrained(_a			).to_dict()

																	config_dict.pop('''image_processor_type'''			)
																	_a      :       Tuple   =						CLIPImageProcessor(**_a			)

																	# save in new folder
																	model_config.save_pretrained(_a			)
																	config.save_pretrained(_a			)

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)

																	# make sure private variable is not incorrectly saved
																	_a      :       Optional[int]   =						json.loads(config.to_json_string()			)
																	self.assertTrue('''_processor_class''' not in dict_as_saved			)

												self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Dict:
												with tempfile.TemporaryDirectory() as tmpdirname:
																	_a      :       Optional[int]   =						Path(_a			) / '''preprocessor_config.json'''
																	json.dump(
																	    {'''image_processor_type''': '''CLIPImageProcessor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)

																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			)
																	self.assertIsInstance(_a			,				_a			)


							def 							__lowercase							(     self			)       ->       Any:
												with self.assertRaisesRegex(
												    _a			,				'''clip-base is not a local folder and is not a valid model identifier'''			):
																	_a      :       Dict   =						AutoImageProcessor.from_pretrained('''clip-base'''			)


							def 							__lowercase							(     self			)       ->       List[Any]:
												with self.assertRaisesRegex(
												    _a			,				R'''aaaaaa is not a valid git identifier \(branch name, tag name or commit id\)'''			):
																	_a      :       List[str]   =						AutoImageProcessor.from_pretrained(_a			,				revision='''aaaaaa'''			)


							def 							__lowercase							(     self			)       ->       Dict:
												with self.assertRaisesRegex(
												    _a			,				'''hf-internal-testing/config-no-model does not appear to have a file named preprocessor_config.json.'''			,				):
																	_a      :       Optional[int]   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/config-no-model'''			)


							def 							__lowercase							(     self			)       ->       Union[str, Any]:
												# If remote code is not set, we will time out when asking whether to load the model.
												with self.assertRaises(_a			):
																	_a      :       str   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/test_dynamic_image_processor'''			)
												# If remote code is disabled, we can't load this config.
												with self.assertRaises(_a			):
																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)

												_a      :       Union[str, Any]   =						AutoImageProcessor.from_pretrained(
												    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
												self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)

												# Test image processor can be reloaded.
												with tempfile.TemporaryDirectory() as tmp_dir:
																	image_processor.save_pretrained(_a			)
																	_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			,				trust_remote_code=_a			)
												self.assertEqual(reloaded_image_processor.__class__.__name__			,				'''NewImageProcessor'''			)


							def 							__lowercase							(     self			)       ->       Dict:
												try:
																	AutoConfig.register('''custom'''			,				_a			)
																	AutoImageProcessor.register(_a			,				_a			)
																	# Trying to register something existing in the Transformers library will raise an error
																	with self.assertRaises(_a			):
																						AutoImageProcessor.register(_a			,				_a			)

																	with tempfile.TemporaryDirectory() as tmpdirname:
																						_a      :       int   =						Path(_a			) / '''preprocessor_config.json'''
																						_a      :       int   =						Path(_a			) / '''config.json'''
																						json.dump(
																						    {'''feature_extractor_type''': '''CLIPFeatureExtractor''', '''processor_class''': '''CLIPProcessor'''}			,				open(_a			,				'''w'''			)			,				)
																						json.dump({'''model_type''': '''clip'''}			,				open(_a			,				'''w'''			)			)

																						_a      :       int   =						CustomImageProcessor.from_pretrained(_a			)

																	# Now that the config is registered, it can be used as any other config with the auto-API
																	with tempfile.TemporaryDirectory() as tmp_dir:
																						image_processor.save_pretrained(_a			)
																						_a      :       Optional[Any]   =						AutoImageProcessor.from_pretrained(_a			)
																						self.assertIsInstance(_a			,				_a			)

												finally:
																	if "custom" in CONFIG_MAPPING._extra_content:
																						del CONFIG_MAPPING._extra_content["custom"]
																	if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
																						del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]




							def 							__lowercase							(     self			)       ->       Union[str, Any]:

												class 							UpperCAmelCase_						(     __lowercase  ):
																			"""simple docstring"""







																			UpperCAmelCase__			:		Tuple             =       True

												try:
																	AutoConfig.register('''custom'''			,				_a			)
																	AutoImageProcessor.register(_a			,				_a			)
																	# If remote code is not set, the default is to use local
																	_a      :       str   =						AutoImageProcessor.from_pretrained('''hf-internal-testing/test_dynamic_image_processor'''			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(image_processor.is_local			)

																	# If remote code is disabled, we load the local one.
																	_a      :       int   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(image_processor.is_local			)

																	# If remote is enabled, we load from the Hub
																	_a      :       Dict   =						AutoImageProcessor.from_pretrained(
																	    '''hf-internal-testing/test_dynamic_image_processor'''			,				trust_remote_code=_a			)
																	self.assertEqual(image_processor.__class__.__name__			,				'''NewImageProcessor'''			)
																	self.assertTrue(not hasattr(_a			,				'''is_local'''			)			)

												finally:
																	if "custom" in CONFIG_MAPPING._extra_content:
																						del CONFIG_MAPPING._extra_content["custom"]
																	if CustomConfig in IMAGE_PROCESSOR_MAPPING._extra_content:
																						del IMAGE_PROCESSOR_MAPPING._extra_content[CustomConfig]



