






import logging
import os
import threading
import time


try:
							import warnings
except ImportError:
							UpperCamelCase								=		None

try:
							import msvcrt
except ImportError:
							UpperCamelCase								=		None

try:
							import fcntl
except ImportError:
							UpperCamelCase								=		None


# Backward compatibility
# ------------------------------------------------
try:
							TimeoutError
except NameError:
							UpperCamelCase								=		OSError


# Data
# ------------------------------------------------
UpperCamelCase								=		[
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

UpperCamelCase								=		"3.0.12"


UpperCamelCase								=		None






def    __magic_name__    (	)     ->    List[Any]:
							global _logger
							_lowercase							:   Optional[int] 							=   _logger or logging.getLogger(__name__				)
							return _logger

class 				lowerCAmelCase_	(		__snake_case			):


				def __init__(					self ,						_lowerCAmelCase      ):
											_lowercase							:   List[Any] 							=   lock_file
											return None


				def __str__(					self      ):
											_lowercase							:   Tuple 							=   F"""The file lock '{self.lock_file}' could not be acquired."""
											return temp




class 				lowerCAmelCase_	:


				def __init__(					self ,						_lowerCAmelCase      ):
											_lowercase							:   Any 							=   lock
											return None


				def __enter__(					self      ):
											return self.lock


				def __exit__(					self ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase      ):
											self.lock.release()
											return None




class 				lowerCAmelCase_	:


				def __init__(					self ,						_lowerCAmelCase ,						_lowerCAmelCase=-1 ,						_lowerCAmelCase=None      ):
											_lowercase							:   Tuple 							=   max_filename_length if max_filename_length is not None else 2_5_5
											# Hash the filename if it's too long
											_lowercase							:   Union[str, Any] 							=   self.hash_filename_if_too_long(_lowerCAmelCase ,						_lowerCAmelCase      )
											# The path to the lock file.
											_lowercase							:   int 							=   lock_file

											# The file descriptor for the *_lock_file* as it is returned by the
											# os.open() function.
											# This file lock is only NOT None, if the object currently holds the
											# lock.
											_lowercase							:   List[Any] 							=   None

											# The default timeout value.
											_lowercase							:   Union[str, Any] 							=   timeout

											# We use this lock primarily for the lock counter.
											_lowercase							:   Any 							=   threading.Lock()

											# The lock counter is used for implementing the nested locking
											# mechanism. Whenever the lock is acquired, the counter is increased and
											# the lock is only released, when this value is 0 again.
											_lowercase							:   Optional[Any] 							=   0
											return None


				@property
				def 						__a					(					self      ):
											return self._lock_file


				@property
				def 						__a					(					self      ):
											return self._timeout


				@timeout.setter
				def 						__a					(					self ,						_lowerCAmelCase      ):
											_lowercase							:   Tuple 							=   float(_lowerCAmelCase      )
											return None


				def 						__a					(					self      ):
											raise NotImplementedError()


				def 						__a					(					self      ):
											raise NotImplementedError()


				@property
				def 						__a					(					self      ):
											return self._lock_file_fd is not None


				def 						__a					(					self ,						_lowerCAmelCase=None ,						_lowerCAmelCase=0.05      ):
											# Use the default timeout, if no timeout is provided.
											if timeout is None:
																		_lowercase							:   Any 							=   self.timeout

											# Increment the number right at the beginning.
											# We can still undo it, if something fails.
											with self._thread_lock:
																		self._lock_counter += 1

											_lowercase							:   Tuple 							=   id(self      )
											_lowercase							:   Union[str, Any] 							=   self._lock_file
											_lowercase							:   Tuple 							=   time.time()
											try:
																		while True:
																									with self._thread_lock:
																																if not self.is_locked:
																																							logger().debug(F"""Attempting to acquire lock {lock_id} on {lock_filename}"""      )
																																							self._acquire()

																									if self.is_locked:
																																logger().debug(F"""Lock {lock_id} acquired on {lock_filename}"""      )
																																break
																									elif timeout >= 0 and time.time() - start_time > timeout:
																																logger().debug(F"""Timeout on acquiring lock {lock_id} on {lock_filename}"""      )
																																raise Timeout(self._lock_file      )
																									else:
																																logger().debug(
																																    F"""Lock {lock_id} not acquired on {lock_filename}, waiting {poll_intervall} seconds ..."""      )
																																time.sleep(_lowerCAmelCase      )
											except:  # noqa
																		# Something did go wrong, so decrement the counter.
																		with self._thread_lock:
																									_lowercase							:   Dict 							=   max(0 ,						self._lock_counter - 1      )

																		raise
											return _Acquire_ReturnProxy(lock=self      )


				def 						__a					(					self ,						_lowerCAmelCase=False      ):
											with self._thread_lock:
																		if self.is_locked:
																									self._lock_counter -= 1

																									if self._lock_counter == 0 or force:
																																_lowercase							:   Tuple 							=   id(self      )
																																_lowercase							:   Optional[Any] 							=   self._lock_file

																																logger().debug(F"""Attempting to release lock {lock_id} on {lock_filename}"""      )
																																self._release()
																																_lowercase							:   Union[str, Any] 							=   0
																																logger().debug(F"""Lock {lock_id} released on {lock_filename}"""      )

											return None


				def __enter__(					self      ):
											self.acquire()
											return self


				def __exit__(					self ,						_lowerCAmelCase ,						_lowerCAmelCase ,						_lowerCAmelCase      ):
											self.release()
											return None


				def __del__(					self      ):
											self.release(force=_lowerCAmelCase      )
											return None


				def 						__a					(					self ,						_lowerCAmelCase ,						_lowerCAmelCase      ):
											_lowercase							:   Optional[Any] 							=   os.path.basename(_lowerCAmelCase      )
											if len(_lowerCAmelCase      ) > max_length and max_length > 0:
																		_lowercase							:   Optional[int] 							=   os.path.dirname(_lowerCAmelCase      )
																		_lowercase							:   int 							=   str(hash(_lowerCAmelCase      )      )
																		_lowercase							:   Union[str, Any] 							=   filename[: max_length - len(_lowerCAmelCase      ) - 8] + '...' + hashed_filename + '.lock'
																		return os.path.join(_lowerCAmelCase ,						_lowerCAmelCase      )
											else:
																		return path




class 				lowerCAmelCase_	(		__snake_case			):


				def __init__(					self ,						_lowerCAmelCase ,						_lowerCAmelCase=-1 ,						_lowerCAmelCase=None      ):
											from .file_utils import relative_to_absolute_path

											super().__init__(_lowerCAmelCase ,						timeout=_lowerCAmelCase ,						max_filename_length=_lowerCAmelCase      )
											_lowercase							:   Optional[Any] 							=   '\\\\?\\' + relative_to_absolute_path(self.lock_file      )


				def 						__a					(					self      ):
											_lowercase							:   str 							=   os.O_RDWR | os.O_CREAT | os.O_TRUNC

											try:
																		_lowercase							:   Dict 							=   os.open(self._lock_file ,						_lowerCAmelCase      )
											except OSError:
																		pass
											else:
																		try:
																									msvcrt.locking(_lowerCAmelCase ,						msvcrt.LK_NBLCK ,						1      )
																		except OSError:
																									os.close(_lowerCAmelCase      )
																		else:
																									_lowercase							:   List[str] 							=   fd
											return None


				def 						__a					(					self      ):
											_lowercase							:   Tuple 							=   self._lock_file_fd
											_lowercase							:   Union[str, Any] 							=   None
											msvcrt.locking(_lowerCAmelCase ,						msvcrt.LK_UNLCK ,						1      )
											os.close(_lowerCAmelCase      )

											try:
																		os.remove(self._lock_file      )
											# Probably another instance of the application
											# that acquired the file lock.
											except OSError:
																		pass
											return None




class 				lowerCAmelCase_	(		__snake_case			):


				def __init__(					self ,						_lowerCAmelCase ,						_lowerCAmelCase=-1 ,						_lowerCAmelCase=None      ):
											_lowercase							:   List[str] 							=   os.statvfs(os.path.dirname(_lowerCAmelCase      )      ).f_namemax
											super().__init__(_lowerCAmelCase ,						timeout=_lowerCAmelCase ,						max_filename_length=_lowerCAmelCase      )


				def 						__a					(					self      ):
											_lowercase							:   List[str] 							=   os.O_RDWR | os.O_CREAT | os.O_TRUNC
											_lowercase							:   List[str] 							=   os.open(self._lock_file ,						_lowerCAmelCase      )

											try:
																		fcntl.flock(_lowerCAmelCase ,						fcntl.LOCK_EX | fcntl.LOCK_NB      )
											except OSError:
																		os.close(_lowerCAmelCase      )
											else:
																		_lowercase							:   Optional[Any] 							=   fd
											return None


				def 						__a					(					self      ):
											# Do not remove the lockfile:
											#
											#   https://github.com/benediktschmitt/py-filelock/issues/31
											#   https://stackoverflow.com/questions/17708885/flock-removing-locked-file-without-race-condition
											_lowercase							:   Tuple 							=   self._lock_file_fd
											_lowercase							:   Optional[Any] 							=   None
											fcntl.flock(_lowerCAmelCase ,						fcntl.LOCK_UN      )
											os.close(_lowerCAmelCase      )
											return None




class 				lowerCAmelCase_	(		__snake_case			):


				def 						__a					(					self      ):
											_lowercase							:   str 							=   os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
											try:
																		_lowercase							:   int 							=   os.open(self._lock_file ,						_lowerCAmelCase      )
											except OSError:
																		pass
											else:
																		_lowercase							:   str 							=   fd
											return None


				def 						__a					(					self      ):
											os.close(self._lock_file_fd      )
											_lowercase							:   Union[str, Any] 							=   None

											try:
																		os.remove(self._lock_file      )
											# The file is already deleted and that's what we want.
											except OSError:
																		pass
											return None




UpperCamelCase								=		None

if msvcrt:
							UpperCamelCase								=		WindowsFileLock
elif fcntl:
							UpperCamelCase								=		UnixFileLock
else:
							UpperCamelCase								=		SoftFileLock

							if warnings is not None:
														warnings.warn("only soft file lock is available")





import gc
import unittest

import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DDPMScheduler,
    PriorTransformer,
    StableUnCLIPPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.stable_unclip_image_normalizer import StableUnCLIPImageNormalizer
from diffusers.utils.testing_utils import enable_full_determinism, load_numpy, require_torch_gpu, slow, torch_device

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import (
    PipelineKarrasSchedulerTesterMixin,
    PipelineLatentTesterMixin,
    PipelineTesterMixin,
    assert_mean_pixel_difference,
)


enable_full_determinism()
class     lowerCAmelCase_     (       __lowercase,  __lowercase,  __lowercase,  unittest.TestCase ):
							UpperCAmelCase				   =     StableUnCLIPPipeline
							UpperCAmelCase				   =     TEXT_TO_IMAGE_PARAMS
							UpperCAmelCase				   =     TEXT_TO_IMAGE_BATCH_PARAMS
							UpperCAmelCase				   =     TEXT_TO_IMAGE_IMAGE_PARAMS
							UpperCAmelCase				   =     TEXT_TO_IMAGE_IMAGE_PARAMS

							# TODO(will) Expected attn_bias.stride(1) == 0 to be true, but got false
							UpperCAmelCase				   =     False







							def  UpperCamelCase_						(   self      :     Optional[int]		):
												_UpperCamelCase			     =     32
												_UpperCamelCase			     =     embedder_hidden_size

												# prior components

												torch.manual_seed(0		)
												_UpperCamelCase			     =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''		)

												torch.manual_seed(0		)
												_UpperCamelCase			     =     CLIPTextModelWithProjection(
												    CLIPTextConfig(
												        bos_token_id=0 ,			eos_token_id=2 ,			hidden_size=_A ,			projection_dim=_A ,			intermediate_size=37 ,			layer_norm_eps=1e-05 ,			num_attention_heads=4 ,			num_hidden_layers=5 ,			pad_token_id=1 ,			vocab_size=1000 ,			)		)

												torch.manual_seed(0		)
												_UpperCamelCase			     =     PriorTransformer(
												    num_attention_heads=2 ,			attention_head_dim=12 ,			embedding_dim=_A ,			num_layers=1 ,			)

												torch.manual_seed(0		)
												_UpperCamelCase			     =     DDPMScheduler(
												    variance_type='''fixed_small_log''' ,			prediction_type='''sample''' ,			num_train_timesteps=1000 ,			clip_sample=_A ,			clip_sample_range=5.0 ,			beta_schedule='''squaredcos_cap_v2''' ,			)

												# regular denoising components

												torch.manual_seed(0		)
												_UpperCamelCase			     =     StableUnCLIPImageNormalizer(embedding_dim=_A		)
												_UpperCamelCase			     =     DDPMScheduler(beta_schedule='''squaredcos_cap_v2'''		)

												torch.manual_seed(0		)
												_UpperCamelCase			     =     CLIPTokenizer.from_pretrained('''hf-internal-testing/tiny-random-clip'''		)

												torch.manual_seed(0		)
												_UpperCamelCase			     =     CLIPTextModel(
												    CLIPTextConfig(
												        bos_token_id=0 ,			eos_token_id=2 ,			hidden_size=_A ,			projection_dim=32 ,			intermediate_size=37 ,			layer_norm_eps=1e-05 ,			num_attention_heads=4 ,			num_hidden_layers=5 ,			pad_token_id=1 ,			vocab_size=1000 ,			)		)

												torch.manual_seed(0		)
												_UpperCamelCase			     =     UNetaDConditionModel(
												    sample_size=32 ,			in_channels=4 ,			out_channels=4 ,			down_block_types=('''CrossAttnDownBlock2D''', '''DownBlock2D''') ,			up_block_types=('''UpBlock2D''', '''CrossAttnUpBlock2D''') ,			block_out_channels=(32, 64) ,			attention_head_dim=(2, 4) ,			class_embed_type='''projection''' ,			projection_class_embeddings_input_dim=embedder_projection_dim * 2 ,			cross_attention_dim=_A ,			layers_per_block=1 ,			upcast_attention=_A ,			use_linear_projection=_A ,			)

												torch.manual_seed(0		)
												_UpperCamelCase			     =     DDIMScheduler(
												    beta_schedule='''scaled_linear''' ,			beta_start=0.0_0085 ,			beta_end=0.012 ,			prediction_type='''v_prediction''' ,			set_alpha_to_one=_A ,			steps_offset=1 ,			)

												torch.manual_seed(0		)
												_UpperCamelCase			     =     AutoencoderKL()

												_UpperCamelCase			     =     {
												    # prior components
												    '''prior_tokenizer''': prior_tokenizer,
												    '''prior_text_encoder''': prior_text_encoder,
												    '''prior''': prior,
												    '''prior_scheduler''': prior_scheduler,
												    # image noising components
												    '''image_normalizer''': image_normalizer,
												    '''image_noising_scheduler''': image_noising_scheduler,
												    # regular denoising components
												    '''tokenizer''': tokenizer,
												    '''text_encoder''': text_encoder,
												    '''unet''': unet,
												    '''scheduler''': scheduler,
												    '''vae''': vae,
												}

												return components







							def  UpperCamelCase_						(   self      :     Dict ,			_A      :     Tuple ,			_A      :     Dict=0		):
												if str(_A		).startswith('''mps'''		):
																	_UpperCamelCase			     =     torch.manual_seed(_A		)
												else:
																	_UpperCamelCase			     =     torch.Generator(device=_A		).manual_seed(_A		)
												_UpperCamelCase			     =     {
												    '''prompt''': '''A painting of a squirrel eating a burger''',
												    '''generator''': generator,
												    '''num_inference_steps''': 2,
												    '''prior_num_inference_steps''': 2,
												    '''output_type''': '''numpy''',
												}
												return inputs







							def  UpperCamelCase_						(   self      :     Dict		):
												_UpperCamelCase			     =     torch_device == '''cpu'''

												self._test_attention_slicing_forward_pass(test_max_difference=_A		)







							def  UpperCamelCase_						(   self      :     List[Any]		):
												_UpperCamelCase			     =     torch_device in ['''cpu''', '''mps''']

												self._test_inference_batch_single_identical(test_max_difference=_A		)




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







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







							def  UpperCamelCase_						(   self      :     List[str]		):
												_UpperCamelCase			     =     load_numpy(
												    '''https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/stable_unclip/stable_unclip_2_1_l_anime_turtle_fp16.npy'''		)

												_UpperCamelCase			     =     StableUnCLIPPipeline.from_pretrained('''fusing/stable-unclip-2-1-l''' ,			torch_dtype=torch.floataa		)
												pipe.to(_A		)
												pipe.set_progress_bar_config(disable=_A		)
												# stable unclip will oom when integration tests are run on a V100,
												# so turn on memory savings
												pipe.enable_attention_slicing()
												pipe.enable_sequential_cpu_offload()

												_UpperCamelCase			     =     torch.Generator(device='''cpu'''		).manual_seed(0		)
												_UpperCamelCase			     =     pipe('''anime turle''' ,			generator=_A ,			output_type='''np'''		)

												_UpperCamelCase			     =     output.images[0]

												assert image.shape == (768, 768, 3)

												assert_mean_pixel_difference(_A ,			_A		)







							def  UpperCamelCase_						(   self      :     Optional[Any]		):
												torch.cuda.empty_cache()
												torch.cuda.reset_max_memory_allocated()
												torch.cuda.reset_peak_memory_stats()

												_UpperCamelCase			     =     StableUnCLIPPipeline.from_pretrained('''fusing/stable-unclip-2-1-l''' ,			torch_dtype=torch.floataa		)
												_UpperCamelCase			     =     pipe.to(_A		)
												pipe.set_progress_bar_config(disable=_A		)
												pipe.enable_attention_slicing()
												pipe.enable_sequential_cpu_offload()

												_UpperCamelCase			     =     pipe(
												    '''anime turtle''' ,			prior_num_inference_steps=2 ,			num_inference_steps=2 ,			output_type='''np''' ,			)

												_UpperCamelCase			     =     torch.cuda.max_memory_allocated()
												# make sure that less than 7 GB is allocated
												assert mem_bytes < 7 * 10**9
