






'''simple docstring'''





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    _lowerCAmelCase			(		unittest.TestCase      ):



				"""simple docstring"""



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






											'''simple docstring'''







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



				@property
				def 					__magic_name__							(      self    :    Optional[int]):






											'''simple docstring'''







											snake_case__	  =       1
											snake_case__	  =       3
											snake_case__	  =       (3_2, 3_2)

											snake_case__	  =       floats_tensor((batch_size, num_channels) + sizes			,		rng=random.Random(0)).to(__snake_case)
											return image



				@property
				def 					__magic_name__							(      self    :    Tuple):






											'''simple docstring'''







											torch.manual_seed(0)
											snake_case__	  =       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 					__magic_name__							(      self    :    List[Any]):






											'''simple docstring'''







											torch.manual_seed(0)
											snake_case__	  =       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 					__magic_name__							(      self    :    str):






											'''simple docstring'''







											torch.manual_seed(0)
											snake_case__	  =       RobertaSeriesConfig(
											    hidden_size=3_2			,		project_dim=3_2			,		intermediate_size=3_7			,		layer_norm_eps=1E-05			,		num_attention_heads=4			,		num_hidden_layers=5			,		pad_token_id=1			,		vocab_size=5_0_0_6			,		)
											return RobertaSeriesModelWithTransformation(__snake_case)



				@property
				def 					__magic_name__							(      self    :    List[Any]):






											'''simple docstring'''







											def extract(*UpperCamelCase__    :    Dict			,		**UpperCamelCase__    :    Optional[Any]):


																		class    _lowerCAmelCase			:



																						"""simple docstring"""



																						def __init__(      self    :    List[str]):






																													'''simple docstring'''







																													snake_case__	  =       torch.ones([0])



																						def 					__magic_name__							(      self    :    Tuple			,		UpperCamelCase__    :    Any):






																													'''simple docstring'''







																													self.pixel_values.to(__snake_case)
																													return self


																		return Out()

											return extract



				def 					__magic_name__							(      self    :    str):






											'''simple docstring'''







											snake_case__	  =       '''cpu'''  # ensure determinism for the device-dependent torch.Generator
											snake_case__	  =       self.dummy_cond_unet
											snake_case__	  =       PNDMScheduler(skip_prk_steps=__snake_case)
											snake_case__	  =       self.dummy_vae
											snake_case__	  =       self.dummy_text_encoder
											snake_case__	  =       XLMRobertaTokenizer.from_pretrained("""hf-internal-testing/tiny-xlm-roberta""")
											snake_case__	  =       7_7

											snake_case__	  =       self.dummy_image.to(__snake_case)
											snake_case__	  =       init_image / 2 + 0.5

											# make sure here that pndm scheduler skips prk
											snake_case__	  =       AltDiffusionImgaImgPipeline(
											    unet=__snake_case			,		scheduler=__snake_case			,		vae=__snake_case			,		text_encoder=__snake_case			,		tokenizer=__snake_case			,		safety_checker=__snake_case			,		feature_extractor=self.dummy_extractor			,		)
											snake_case__	  =       VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor			,		do_normalize=__snake_case)
											snake_case__	  =       alt_pipe.to(__snake_case)
											alt_pipe.set_progress_bar_config(disable=__snake_case)

											snake_case__	  =       '''A painting of a squirrel eating a burger'''
											snake_case__	  =       torch.Generator(device=__snake_case).manual_seed(0)
											snake_case__	  =       alt_pipe(
											    [prompt]			,		generator=__snake_case			,		guidance_scale=6.0			,		num_inference_steps=2			,		output_type="""np"""			,		image=__snake_case			,		)

											snake_case__	  =       output.images

											snake_case__	  =       torch.Generator(device=__snake_case).manual_seed(0)
											snake_case__	  =       alt_pipe(
											    [prompt]			,		generator=__snake_case			,		guidance_scale=6.0			,		num_inference_steps=2			,		output_type="""np"""			,		image=__snake_case			,		return_dict=__snake_case			,		)[0]

											snake_case__	  =       image[0, -3:, -3:, -1]
											snake_case__	  =       image_from_tuple[0, -3:, -3:, -1]

											assert image.shape == (1, 3_2, 3_2, 3)
											snake_case__	  =       np.array([0.44_27, 0.37_31, 0.42_49, 0.49_41, 0.45_46, 0.41_48, 0.41_93, 0.46_66, 0.44_99])

											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 					__magic_name__							(      self    :    Optional[Any]):






											'''simple docstring'''







											snake_case__	  =       self.dummy_cond_unet
											snake_case__	  =       PNDMScheduler(skip_prk_steps=__snake_case)
											snake_case__	  =       self.dummy_vae
											snake_case__	  =       self.dummy_text_encoder
											snake_case__	  =       XLMRobertaTokenizer.from_pretrained("""hf-internal-testing/tiny-xlm-roberta""")
											snake_case__	  =       7_7

											snake_case__	  =       self.dummy_image.to(__snake_case)

											# put models in fp16
											snake_case__	  =       unet.half()
											snake_case__	  =       vae.half()
											snake_case__	  =       bert.half()

											# make sure here that pndm scheduler skips prk
											snake_case__	  =       AltDiffusionImgaImgPipeline(
											    unet=__snake_case			,		scheduler=__snake_case			,		vae=__snake_case			,		text_encoder=__snake_case			,		tokenizer=__snake_case			,		safety_checker=__snake_case			,		feature_extractor=self.dummy_extractor			,		)
											snake_case__	  =       VaeImageProcessor(vae_scale_factor=alt_pipe.vae_scale_factor			,		do_normalize=__snake_case)
											snake_case__	  =       alt_pipe.to(__snake_case)
											alt_pipe.set_progress_bar_config(disable=__snake_case)

											snake_case__	  =       '''A painting of a squirrel eating a burger'''
											snake_case__	  =       torch.manual_seed(0)
											snake_case__	  =       alt_pipe(
											    [prompt]			,		generator=__snake_case			,		num_inference_steps=2			,		output_type="""np"""			,		image=__snake_case			,		).images

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



				@unittest.skipIf(torch_device != """cuda"""			,		"""This test requires a GPU""")
				def 					__magic_name__							(      self    :    List[Any]):






											'''simple docstring'''







											snake_case__	  =       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
											snake_case__	  =       init_image.resize((7_6_0, 5_0_4))

											snake_case__	  =       '''BAAI/AltDiffusion'''
											snake_case__	  =       AltDiffusionImgaImgPipeline.from_pretrained(
											    __snake_case			,		safety_checker=__snake_case			,		)
											pipe.to(__snake_case)
											pipe.set_progress_bar_config(disable=__snake_case)
											pipe.enable_attention_slicing()

											snake_case__	  =       '''A fantasy landscape, trending on artstation'''

											snake_case__	  =       torch.manual_seed(0)
											snake_case__	  =       pipe(
											    prompt=__snake_case			,		image=__snake_case			,		strength=0.75			,		guidance_scale=7.5			,		generator=__snake_case			,		output_type="""np"""			,		)
											snake_case__	  =       output.images[0]

											snake_case__	  =       image[2_5_5:2_5_8, 3_8_3:3_8_6, -1]

											assert image.shape == (5_0_4, 7_6_0, 3)
											snake_case__	  =       np.array([0.93_58, 0.93_97, 0.95_99, 0.99_01, 1.00_00, 1.00_00, 0.98_82, 1.00_00, 1.00_00])

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


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



				"""simple docstring"""



				def 					__magic_name__							(      self    :    Optional[Any]):






											'''simple docstring'''







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



				def 					__magic_name__							(      self    :    int):






											'''simple docstring'''







											snake_case__	  =       load_image(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
											    """/img2img/sketch-mountains-input.jpg""")
											snake_case__	  =       init_image.resize((7_6_8, 5_1_2))
											snake_case__	  =       load_numpy(
											    """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/img2img/fantasy_landscape_alt.npy""")

											snake_case__	  =       '''BAAI/AltDiffusion'''
											snake_case__	  =       AltDiffusionImgaImgPipeline.from_pretrained(
											    __snake_case			,		safety_checker=__snake_case			,		)
											pipe.to(__snake_case)
											pipe.set_progress_bar_config(disable=__snake_case)
											pipe.enable_attention_slicing()

											snake_case__	  =       '''A fantasy landscape, trending on artstation'''

											snake_case__	  =       torch.manual_seed(0)
											snake_case__	  =       pipe(
											    prompt=__snake_case			,		image=__snake_case			,		strength=0.75			,		guidance_scale=7.5			,		generator=__snake_case			,		output_type="""np"""			,		)
											snake_case__	  =       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 os
import time

import pytest

from datasets.utils.filelock import FileLock, Timeout

def _UpperCAmelCase   (	a						:	List[Any] ):
							snake_case__	  =       FileLock(str(tmpdir / """foo.lock""" ) )
							snake_case__	  =       FileLock(str(tmpdir / """foo.lock""" ) )
							snake_case__	  =       0.01
							with locka.acquire():
														with pytest.raises(a ):
																					snake_case__	  =       time.time()
																					locka.acquire(a )
																					assert time.time() - _start > timeout



def _UpperCAmelCase   (	a						:	str ):
							snake_case__	  =       """a""" * 1000 + """.lock"""
							snake_case__	  =       FileLock(str(tmpdir / filename ) )
							assert locka._lock_file.endswith(""".lock""" )
							assert not locka._lock_file.endswith(a )
							assert len(os.path.basename(locka._lock_file ) ) <= 255
							snake_case__	  =       FileLock(tmpdir / filename )
							with locka.acquire():
														with pytest.raises(a ):
																					locka.acquire(0 )
