
"""simple docstring"""


import logging
import os
import threading
import time


try:
							import warnings
except ImportError:
							lowerCAmelCase								=						None

try:
							import msvcrt
except ImportError:
							lowerCAmelCase								=						None

try:
							import fcntl
except ImportError:
							lowerCAmelCase								=						None


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


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

lowerCAmelCase								=						"3.0.12"


lowerCAmelCase								=						None






def    lowerCAmelCase_   (		)   ->Union[str, Any]:
			global _logger
			lowerCamelCase__	:     Tuple              =_logger or logging.getLogger(__name__			)
			return _logger

class     A_	(     UpperCamelCase__   ):



							"""simple docstring"""






							def __init__(  self  :List[str]     ,							lowerCamelCase_  :Any       ):





										"""simple docstring"""

										lowerCamelCase__	:     str              =lock_file
										return None





							def __str__(  self  :List[str]       ):





										"""simple docstring"""

										lowerCamelCase__	:     Dict              =f"""The file lock '{self.lock_file}' could not be acquired."""
										return temp






class     A_	:



							"""simple docstring"""






							def __init__(  self  :str     ,							lowerCamelCase_  :str       ):





										"""simple docstring"""

										lowerCamelCase__	:     str              =lock
										return None





							def __enter__(  self  :Dict       ):





										"""simple docstring"""

										return self.lock





							def __exit__(  self  :int     ,							lowerCamelCase_  :int     ,							lowerCamelCase_  :List[str]     ,							lowerCamelCase_  :Any       ):





										"""simple docstring"""

										self.lock.release()
										return None






class     A_	:



							"""simple docstring"""






							def __init__(  self  :List[str]     ,							lowerCamelCase_  :Dict     ,							lowerCamelCase_  :Optional[int]=-1     ,							lowerCamelCase_  :Optional[int]=None       ):





										"""simple docstring"""

										lowerCamelCase__	:     Union[str, Any]              =max_filename_length if max_filename_length is not None else 255
										# Hash the filename if it's too long
										lowerCamelCase__	:     int              =self.hash_filename_if_too_long(lowercase_     ,							lowercase_       )
										# The path to the lock file.
										lowerCamelCase__	:     Union[str, Any]              =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.
										lowerCamelCase__	:     Dict              =None

										# The default timeout value.
										lowerCamelCase__	:     List[Any]              =timeout

										# We use this lock primarily for the lock counter.
										lowerCamelCase__	:     int              =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.
										lowerCamelCase__	:     List[str]              =0
										return None





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





										"""simple docstring"""

										return self._lock_file





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





										"""simple docstring"""

										return self._timeout





							@timeout.setter
							def       UpperCAmelCase__							(  self  :List[Any]     ,							lowerCamelCase_  :str       ):





										"""simple docstring"""

										lowerCamelCase__	:     Union[str, Any]              =float(lowercase_       )
										return None





							def       UpperCAmelCase__							(  self  :Dict       ):





										"""simple docstring"""

										raise NotImplementedError()





							def       UpperCAmelCase__							(  self  :Optional[Any]       ):





										"""simple docstring"""

										raise NotImplementedError()





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





										"""simple docstring"""

										return self._lock_file_fd is not None





							def       UpperCAmelCase__							(  self  :List[str]     ,							lowerCamelCase_  :List[str]=None     ,							lowerCamelCase_  :Tuple=0.05       ):





										"""simple docstring"""

										if timeout is None:
													lowerCamelCase__	:     int              =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

										lowerCamelCase__	:     Any              =id(self       )
										lowerCamelCase__	:     Tuple              =self._lock_file
										lowerCamelCase__	:     Optional[int]              =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(lowercase_       )
										except:  # noqa
													# Something did go wrong, so decrement the counter.
													with self._thread_lock:
																lowerCamelCase__	:     Dict              =max(0     ,							self._lock_counter - 1       )

													raise
										return _Acquire_ReturnProxy(lock=self       )





							def       UpperCAmelCase__							(  self  :Tuple     ,							lowerCamelCase_  :Optional[int]=False       ):





										"""simple docstring"""

										with self._thread_lock:
													if self.is_locked:
																self._lock_counter -= 1

																if self._lock_counter == 0 or force:
																			lowerCamelCase__	:     List[str]              =id(self       )
																			lowerCamelCase__	:     List[Any]              =self._lock_file

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

										return None





							def __enter__(  self  :int       ):





										"""simple docstring"""

										self.acquire()
										return self





							def __exit__(  self  :Tuple     ,							lowerCamelCase_  :Tuple     ,							lowerCamelCase_  :str     ,							lowerCamelCase_  :Any       ):





										"""simple docstring"""

										self.release()
										return None





							def __del__(  self  :str       ):





										"""simple docstring"""

										self.release(force=lowercase_       )
										return None





							def       UpperCAmelCase__							(  self  :str     ,							lowerCamelCase_  :str     ,							lowerCamelCase_  :int       ):





										"""simple docstring"""

										lowerCamelCase__	:     Dict              =os.path.basename(lowercase_       )
										if len(lowercase_       ) > max_length and max_length > 0:
													lowerCamelCase__	:     List[Any]              =os.path.dirname(lowercase_       )
													lowerCamelCase__	:     Dict              =str(hash(lowercase_       )       )
													lowerCamelCase__	:     Any              =filename[: max_length - len(lowercase_       ) - 8] + '...' + hashed_filename + '.lock'
													return os.path.join(lowercase_     ,							lowercase_       )
										else:
													return path






class     A_	(     UpperCamelCase__   ):



							"""simple docstring"""






							def __init__(  self  :Dict     ,							lowerCamelCase_  :Union[str, Any]     ,							lowerCamelCase_  :Optional[Any]=-1     ,							lowerCamelCase_  :Dict=None       ):





										"""simple docstring"""

										from .file_utils import relative_to_absolute_path

										super().__init__(lowercase_     ,							timeout=lowercase_     ,							max_filename_length=lowercase_       )
										lowerCamelCase__	:     Any              ='\\\\?\\' + relative_to_absolute_path(self.lock_file       )





							def       UpperCAmelCase__							(  self  :Any       ):





										"""simple docstring"""

										lowerCamelCase__	:     Union[str, Any]              =os.O_RDWR | os.O_CREAT | os.O_TRUNC

										try:
													lowerCamelCase__	:     int              =os.open(self._lock_file     ,							lowercase_       )
										except OSError:
													pass
										else:
													try:
																msvcrt.locking(lowercase_     ,							msvcrt.LK_NBLCK     ,							1       )
													except OSError:
																os.close(lowercase_       )
													else:
																lowerCamelCase__	:     List[Any]              =fd
										return None





							def       UpperCAmelCase__							(  self  :int       ):





										"""simple docstring"""

										lowerCamelCase__	:     str              =self._lock_file_fd
										lowerCamelCase__	:     Union[str, Any]              =None
										msvcrt.locking(lowercase_     ,							msvcrt.LK_UNLCK     ,							1       )
										os.close(lowercase_       )

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






class     A_	(     UpperCamelCase__   ):



							"""simple docstring"""






							def __init__(  self  :Optional[Any]     ,							lowerCamelCase_  :Tuple     ,							lowerCamelCase_  :List[str]=-1     ,							lowerCamelCase_  :List[Any]=None       ):





										"""simple docstring"""

										lowerCamelCase__	:     Optional[Any]              =os.statvfs(os.path.dirname(lowercase_       )       ).f_namemax
										super().__init__(lowercase_     ,							timeout=lowercase_     ,							max_filename_length=lowercase_       )





							def       UpperCAmelCase__							(  self  :Tuple       ):





										"""simple docstring"""

										lowerCamelCase__	:     Dict              =os.O_RDWR | os.O_CREAT | os.O_TRUNC
										lowerCamelCase__	:     int              =os.open(self._lock_file     ,							lowercase_       )

										try:
													fcntl.flock(lowercase_     ,							fcntl.LOCK_EX | fcntl.LOCK_NB       )
										except OSError:
													os.close(lowercase_       )
										else:
													lowerCamelCase__	:     List[Any]              =fd
										return None





							def       UpperCAmelCase__							(  self  :List[Any]       ):





										"""simple docstring"""

										lowerCamelCase__	:     Tuple              =self._lock_file_fd
										lowerCamelCase__	:     int              =None
										fcntl.flock(lowercase_     ,							fcntl.LOCK_UN       )
										os.close(lowercase_       )
										return None






class     A_	(     UpperCamelCase__   ):



							"""simple docstring"""






							def       UpperCAmelCase__							(  self  :str       ):





										"""simple docstring"""

										lowerCamelCase__	:     List[str]              =os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
										try:
													lowerCamelCase__	:     Optional[Any]              =os.open(self._lock_file     ,							lowercase_       )
										except OSError:
													pass
										else:
													lowerCamelCase__	:     Any              =fd
										return None





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





										"""simple docstring"""

										os.close(self._lock_file_fd       )
										lowerCamelCase__	:     Dict              =None

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






lowerCAmelCase								=						None

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

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






'''simple docstring'''

import gc
import random
import unittest

import numpy as np
import torch

from diffusers import DDIMScheduler, KandinskyVaaPipeline, KandinskyVaaPriorPipeline, UNetaDConditionModel, VQModel
from diffusers.utils import floats_tensor, load_numpy, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..test_pipelines_common import PipelineTesterMixin, assert_mean_pixel_difference


enable_full_determinism()






class    UpperCAmelCase	(       UpperCamelCase__			,       unittest.TestCase				):
 __lowercase   		=				KandinskyVaaPipeline
 __lowercase   		=				[
     """image_embeds""",
     """negative_image_embeds""",
 ]
 __lowercase   		=				["""image_embeds""", """negative_image_embeds"""]
 __lowercase   		=				[
     """generator""",
     """height""",
     """width""",
     """latents""",
     """guidance_scale""",
     """num_inference_steps""",
     """return_dict""",
     """guidance_scale""",
     """num_images_per_prompt""",
     """output_type""",
     """return_dict""",
 ]
 __lowercase   		=				False
 @property
 def      UpperCAmelCase_							(				self      :Optional[Any]							)->							str:
     return 32
 @property
 def      UpperCAmelCase_							(				self      :int							)->							List[Any]:
     return 32
 @property
 def      UpperCAmelCase_							(				self      :Optional[Any]							)->							Dict:
     return self.time_input_dim
 @property
 def      UpperCAmelCase_							(				self      :Any							)->							Union[str, Any]:
     return self.time_input_dim * 4
 @property
 def      UpperCAmelCase_							(				self      :Any							)->							List[Any]:
     return 1_00
 @property
 def      UpperCAmelCase_							(				self      :Union[str, Any]							)->							Union[str, Any]:
     torch.manual_seed(0							)

     A__				=			{
         "in_channels": 4,
         # Out channels is double in channels because predicts mean and variance
         "out_channels": 8,
         "addition_embed_type": "image",
         "down_block_types": ("ResnetDownsampleBlock2D", "SimpleCrossAttnDownBlock2D"),
         "up_block_types": ("SimpleCrossAttnUpBlock2D", "ResnetUpsampleBlock2D"),
         "mid_block_type": "UNetMidBlock2DSimpleCrossAttn",
         "block_out_channels": (self.block_out_channels_a, self.block_out_channels_a * 2),
         "layers_per_block": 1,
         "encoder_hid_dim": self.text_embedder_hidden_size,
         "encoder_hid_dim_type": "image_proj",
         "cross_attention_dim": self.cross_attention_dim,
         "attention_head_dim": 4,
         "resnet_time_scale_shift": "scale_shift",
         "class_embed_type": None,
     }

     A__				=			UNetaDConditionModel(**lowercase_							)
     return model
 @property
 def      UpperCAmelCase_							(				self      :Any							)->							Optional[Any]:
     return {
         "block_out_channels": [32, 64],
         "down_block_types": ["DownEncoderBlock2D", "AttnDownEncoderBlock2D"],
         "in_channels": 3,
         "latent_channels": 4,
         "layers_per_block": 1,
         "norm_num_groups": 8,
         "norm_type": "spatial",
         "num_vq_embeddings": 12,
         "out_channels": 3,
         "up_block_types": [
             "AttnUpDecoderBlock2D",
             "UpDecoderBlock2D",
         ],
         "vq_embed_dim": 4,
     }
 @property
 def      UpperCAmelCase_							(				self      :Dict							)->							Union[str, Any]:
     torch.manual_seed(0							)
     A__				=			VQModel(**self.dummy_movq_kwargs							)
     return model
 def      UpperCAmelCase_							(				self      :Dict							)->							Optional[Any]:
     A__				=			self.dummy_unet
     A__				=			self.dummy_movq

     A__				=			DDIMScheduler(
         num_train_timesteps=10_00						,       beta_schedule="linear"						,       beta_start=0.0_0_0_8_5						,       beta_end=0.0_1_2						,       clip_sample=lowercase_						,       set_alpha_to_one=lowercase_						,       steps_offset=1						,       prediction_type="epsilon"						,       thresholding=lowercase_						,       )

     A__				=			{
         "unet": unet,
         "scheduler": scheduler,
         "movq": movq,
     }
     return components
 def      UpperCAmelCase_							(				self      :Dict						,       lowercase_      :int						,       lowercase_      :Union[str, Any]=0							)->							Dict:
     A__				=			floats_tensor((1, self.text_embedder_hidden_size)						,       rng=random.Random(lowercase_							)							).to(lowercase_							)
     A__				=			floats_tensor((1, self.text_embedder_hidden_size)						,       rng=random.Random(seed + 1							)							).to(
         lowercase_							)
     if str(lowercase_							).startswith("mps"							):
         A__				=			torch.manual_seed(lowercase_							)
     else:
         A__				=			torch.Generator(device=lowercase_							).manual_seed(lowercase_							)
     A__				=			{
         "image_embeds": image_embeds,
         "negative_image_embeds": negative_image_embeds,
         "generator": generator,
         "height": 64,
         "width": 64,
         "guidance_scale": 4.0,
         "num_inference_steps": 2,
         "output_type": "np",
     }
     return inputs



 def      UpperCAmelCase_							(				self      :Optional[Any]							)->							str:
     A__				=			"cpu"

     A__				=			self.get_dummy_components()

     A__				=			self.pipeline_class(**lowercase_							)
     A__				=			pipe.to(lowercase_							)

     pipe.set_progress_bar_config(disable=lowercase_							)

     A__				=			pipe(**self.get_dummy_inputs(lowercase_							)							)
     A__				=			output.images

     A__				=			pipe(
         **self.get_dummy_inputs(lowercase_							)						,       return_dict=lowercase_						,       )[0]

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

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

     A__				=			np.array(
         [0.6_2_3_7_9_7_6, 1.0, 0.3_6_4_4_1_3_3_2, 1.0, 0.7_0_6_3_9_6_3_4, 0.2_9_8_7_7_1_8_6, 0.8_5_6_5_2_1_2_5, 0.5_2_1_6_8_4_3, 0.5_4_4_5_4_0_4_6]							)

     assert (
         np.abs(image_slice.flatten() - expected_slice							).max() < 1E-2
     ), F" expected_slice {expected_slice}, but got {image_slice.flatten()}"

     assert (
         np.abs(image_from_tuple_slice.flatten() - expected_slice							).max() < 1E-2
     ), F" expected_slice {expected_slice}, but got {image_from_tuple_slice.flatten()}"






@slow
@require_torch_gpu
class    UpperCAmelCase	(       unittest.TestCase				):
 def      UpperCAmelCase_							(				self      :Dict							)->							str:
     # clean up the VRAM after each test
     super().tearDown()
     gc.collect()
     torch.cuda.empty_cache()



 def      UpperCAmelCase_							(				self      :Any							)->							List[Any]:
     A__				=			load_numpy(
         "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
         "/kandinskyv22/kandinskyv22_text2img_cat_fp16.npy"							)

     A__				=			KandinskyVaaPriorPipeline.from_pretrained(
         "kandinsky-community/kandinsky-2-2-prior"						,       torch_dtype=torch.floataa							)
     pipe_prior.to(lowercase_							)

     A__				=			KandinskyVaaPipeline.from_pretrained(
         "kandinsky-community/kandinsky-2-2-decoder"						,       torch_dtype=torch.floataa							)
     A__				=			pipeline.to(lowercase_							)
     pipeline.set_progress_bar_config(disable=lowercase_							)

     A__				=			"red cat, 4k photo"

     A__				=			torch.Generator(device="cuda"							).manual_seed(0							)
     A__,	A__				=			pipe_prior(
         lowercase_						,       generator=lowercase_						,       num_inference_steps=5						,       negative_prompt=""						,       ).to_tuple()

     A__				=			torch.Generator(device="cuda"							).manual_seed(0							)
     A__				=			pipeline(
         image_embeds=lowercase_						,       negative_image_embeds=lowercase_						,       generator=lowercase_						,       num_inference_steps=1_00						,       output_type="np"						,       )

     A__				=			output.images[0]

     assert image.shape == (5_12, 5_12, 3)

     assert_mean_pixel_difference(lowercase_						,       lowercase_							)



