
"""simple docstring"""



import logging
import os
import threading
import time


try:
 import warnings
except ImportError:
 a					:	Optional[int]							=		None

try:
 import msvcrt
except ImportError:
 a					:	Optional[Any]							=		None

try:
 import fcntl
except ImportError:
 a					:	List[str]							=		None


# Backward compatibility
# ------------------------------------------------
try:
 TimeoutError
except NameError:
 a					:	Optional[Any]							=		OSError


# Data
# ------------------------------------------------
a					:	List[Any]							=		[
    '''Timeout''',
    '''BaseFileLock''',
    '''WindowsFileLock''',
    '''UnixFileLock''',
    '''SoftFileLock''',
    '''FileLock''',
]

a					:	int							=		'''3.0.12'''


a					:	Dict							=		None


def  _UpperCamelCase	(						)    ->  Optional[int]:





    """simple docstring"""

    global _logger
    _UpperCAmelCase          =   _logger or logging.getLogger(__name__      )
    return _logger




class 		a_						( _UpperCAmelCase	):







    def __init__(			self			:       Any     ,					__UpperCamelCase			:       Dict     )		->Tuple:
        _UpperCAmelCase          =   lock_file
        return None







    def __str__(			self			:       List[str]     )		->Tuple:
        _UpperCAmelCase          =   f"""The file lock '{self.lock_file}' could not be acquired."""
        return temp







class 		a_						:







    def __init__(			self			:       Tuple     ,					__UpperCamelCase			:       Any     )		->Any:
        _UpperCAmelCase          =   lock
        return None







    def __enter__(			self			:       Any     )		->str:
        return self.lock







    def __exit__(			self			:       List[Any]     ,					__UpperCamelCase			:       int     ,					__UpperCamelCase			:       List[str]     ,					__UpperCamelCase			:       Optional[int]     )		->List[str]:
        self.lock.release()
        return None







class 		a_						:







    def __init__(			self			:       Optional[Any]     ,					__UpperCamelCase			:       Optional[int]     ,					__UpperCamelCase			:       Dict=-1     ,					__UpperCamelCase			:       Union[str, Any]=None     )		->Optional[Any]:
        _UpperCAmelCase          =   max_filename_length if max_filename_length is not None else 2_55
        # Hash the filename if it's too long
        _UpperCAmelCase          =   self.hash_filename_if_too_long(__UpperCamelCase     ,					__UpperCamelCase     )
        # The path to the lock file.
        _UpperCAmelCase          =   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.
        _UpperCAmelCase          =   None

        # The default timeout value.
        _UpperCAmelCase          =   timeout

        # We use this lock primarily for the lock counter.
        _UpperCAmelCase          =   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.
        _UpperCAmelCase          =   0
        return None







    @property
    def      _snake_case					(			self			:       Union[str, Any]     )		->Optional[Any]:
        return self._lock_file







    @property
    def      _snake_case					(			self			:       Tuple     )		->Any:
        return self._timeout







    @timeout.setter
    def      _snake_case					(			self			:       Union[str, Any]     ,					__UpperCamelCase			:       Any     )		->int:
        _UpperCAmelCase          =   float(__UpperCamelCase     )
        return None







    def      _snake_case					(			self			:       Optional[int]     )		->List[Any]:
        raise NotImplementedError()







    def      _snake_case					(			self			:       str     )		->str:
        raise NotImplementedError()







    @property
    def      _snake_case					(			self			:       Optional[int]     )		->int:
        return self._lock_file_fd is not None







    def      _snake_case					(			self			:       Any     ,					__UpperCamelCase			:       List[Any]=None     ,					__UpperCamelCase			:       int=0.0_5     )		->Tuple:






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

        _UpperCAmelCase          =   id(self     )
        _UpperCAmelCase          =   self._lock_file
        _UpperCAmelCase          =   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(__UpperCamelCase     )
        except:  # noqa
            # Something did go wrong, so decrement the counter.
            with self._thread_lock:
                _UpperCAmelCase          =   max(0     ,					self._lock_counter - 1     )

            raise
        return _Acquire_ReturnProxy(lock=self     )







    def      _snake_case					(			self			:       str     ,					__UpperCamelCase			:       Tuple=False     )		->Dict:






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

                if self._lock_counter == 0 or force:
                    _UpperCAmelCase          =   id(self     )
                    _UpperCAmelCase          =   self._lock_file

                    logger().debug(f"""Attempting to release lock {lock_id} on {lock_filename}"""     )
                    self._release()
                    _UpperCAmelCase          =   0
                    logger().debug(f"""Lock {lock_id} released on {lock_filename}"""     )

        return None







    def __enter__(			self			:       str     )		->Dict:
        self.acquire()
        return self







    def __exit__(			self			:       int     ,					__UpperCamelCase			:       Dict     ,					__UpperCamelCase			:       Dict     ,					__UpperCamelCase			:       Optional[int]     )		->Optional[Any]:
        self.release()
        return None







    def __del__(			self			:       Any     )		->Tuple:
        self.release(force=__UpperCamelCase     )
        return None







    def      _snake_case					(			self			:       int     ,					__UpperCamelCase			:       str     ,					__UpperCamelCase			:       int     )		->str:
        _UpperCAmelCase          =   os.path.basename(__UpperCamelCase     )
        if len(__UpperCamelCase     ) > max_length and max_length > 0:
            _UpperCAmelCase          =   os.path.dirname(__UpperCamelCase     )
            _UpperCAmelCase          =   str(hash(__UpperCamelCase     )     )
            _UpperCAmelCase          =   filename[: max_length - len(__UpperCamelCase     ) - 8] + """...""" + hashed_filename + """.lock"""
            return os.path.join(__UpperCamelCase     ,					__UpperCamelCase     )
        else:
            return path







class 		a_						( _UpperCAmelCase	):







    def __init__(			self			:       Optional[Any]     ,					__UpperCamelCase			:       Optional[Any]     ,					__UpperCamelCase			:       Tuple=-1     ,					__UpperCamelCase			:       Dict=None     )		->int:
        from .file_utils import relative_to_absolute_path

        super().__init__(__UpperCamelCase     ,					timeout=__UpperCamelCase     ,					max_filename_length=__UpperCamelCase     )
        _UpperCAmelCase          =   """\\\\?\\""" + relative_to_absolute_path(self.lock_file     )







    def      _snake_case					(			self			:       List[str]     )		->List[Any]:
        _UpperCAmelCase          =   os.O_RDWR | os.O_CREAT | os.O_TRUNC

        try:
            _UpperCAmelCase          =   os.open(self._lock_file     ,					__UpperCamelCase     )
        except OSError:
            pass
        else:
            try:
                msvcrt.locking(__UpperCamelCase     ,					msvcrt.LK_NBLCK     ,					1     )
            except OSError:
                os.close(__UpperCamelCase     )
            else:
                _UpperCAmelCase          =   fd
        return None







    def      _snake_case					(			self			:       Tuple     )		->Tuple:
        _UpperCAmelCase          =   self._lock_file_fd
        _UpperCAmelCase          =   None
        msvcrt.locking(__UpperCamelCase     ,					msvcrt.LK_UNLCK     ,					1     )
        os.close(__UpperCamelCase     )

        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	):







    def __init__(			self			:       Any     ,					__UpperCamelCase			:       Any     ,					__UpperCamelCase			:       Tuple=-1     ,					__UpperCamelCase			:       Union[str, Any]=None     )		->Any:
        _UpperCAmelCase          =   os.statvfs(os.path.dirname(__UpperCamelCase     )     ).f_namemax
        super().__init__(__UpperCamelCase     ,					timeout=__UpperCamelCase     ,					max_filename_length=__UpperCamelCase     )







    def      _snake_case					(			self			:       Optional[Any]     )		->Tuple:
        _UpperCAmelCase          =   os.O_RDWR | os.O_CREAT | os.O_TRUNC
        _UpperCAmelCase          =   os.open(self._lock_file     ,					__UpperCamelCase     )

        try:
            fcntl.flock(__UpperCamelCase     ,					fcntl.LOCK_EX | fcntl.LOCK_NB     )
        except OSError:
            os.close(__UpperCamelCase     )
        else:
            _UpperCAmelCase          =   fd
        return None







    def      _snake_case					(			self			:       List[str]     )		->Optional[Any]:
        _UpperCAmelCase          =   self._lock_file_fd
        _UpperCAmelCase          =   None
        fcntl.flock(__UpperCamelCase     ,					fcntl.LOCK_UN     )
        os.close(__UpperCamelCase     )
        return None







class 		a_						( _UpperCAmelCase	):







    def      _snake_case					(			self			:       List[Any]     )		->Optional[Any]:
        _UpperCAmelCase          =   os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
        try:
            _UpperCAmelCase          =   os.open(self._lock_file     ,					__UpperCamelCase     )
        except OSError:
            pass
        else:
            _UpperCAmelCase          =   fd
        return None







    def      _snake_case					(			self			:       str     )		->List[str]:
        os.close(self._lock_file_fd     )
        _UpperCAmelCase          =   None

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







a					:	str							=		None

if msvcrt:
 a					:	Optional[int]							=		WindowsFileLock
elif fcntl:
 a					:	Tuple							=		UnixFileLock
else:
 a					:	Any							=		SoftFileLock

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

"""simple docstring"""



import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel

from diffusers import DDIMScheduler, LDMPipeline, UNetaDModel, VQModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch, slow, torch_device


enable_full_determinism()




class 		a_						( unittest.TestCase	):







    @property
    def      _snake_case					(			self			:       Dict     )		->int:






        '''simple docstring'''






        torch.manual_seed(0     )
        _UpperCAmelCase          =   UNetaDModel(
            block_out_channels=(32, 64)     ,					layers_per_block=2     ,					sample_size=32     ,					in_channels=3     ,					out_channels=3     ,					down_block_types=("""DownBlock2D""", """AttnDownBlock2D""")     ,					up_block_types=("""AttnUpBlock2D""", """UpBlock2D""")     ,					)
        return model







    @property
    def      _snake_case					(			self			:       Optional[Any]     )		->str:






        '''simple docstring'''






        torch.manual_seed(0     )
        _UpperCAmelCase          =   VQModel(
            block_out_channels=[32, 64]     ,					in_channels=3     ,					out_channels=3     ,					down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]     ,					up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]     ,					latent_channels=3     ,					)
        return model







    @property
    def      _snake_case					(			self			:       List[Any]     )		->Optional[int]:






        '''simple docstring'''






        torch.manual_seed(0     )
        _UpperCAmelCase          =   CLIPTextConfig(
            bos_token_id=0     ,					eos_token_id=2     ,					hidden_size=32     ,					intermediate_size=37     ,					layer_norm_eps=1e-05     ,					num_attention_heads=4     ,					num_hidden_layers=5     ,					pad_token_id=1     ,					vocab_size=10_00     ,					)
        return CLIPTextModel(__UpperCamelCase     )







    def      _snake_case					(			self			:       List[str]     )		->Optional[Any]:






        '''simple docstring'''






        _UpperCAmelCase          =   self.dummy_uncond_unet
        _UpperCAmelCase          =   DDIMScheduler()
        _UpperCAmelCase          =   self.dummy_vq_model

        _UpperCAmelCase          =   LDMPipeline(unet=__UpperCamelCase     ,					vqvae=__UpperCamelCase     ,					scheduler=__UpperCamelCase     )
        ldm.to(__UpperCamelCase     )
        ldm.set_progress_bar_config(disable=__UpperCamelCase     )

        _UpperCAmelCase          =   torch.manual_seed(0     )
        _UpperCAmelCase          =   ldm(generator=__UpperCamelCase     ,					num_inference_steps=2     ,					output_type="""numpy"""     ).images

        _UpperCAmelCase          =   torch.manual_seed(0     )
        _UpperCAmelCase          =   ldm(generator=__UpperCamelCase     ,					num_inference_steps=2     ,					output_type="""numpy"""     ,					return_dict=__UpperCamelCase     )[0]

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

        assert image.shape == (1, 64, 64, 3)
        _UpperCAmelCase          =   np.array([0.8_5_1_2, 0.8_1_8, 0.6_4_1_1, 0.6_8_0_8, 0.4_4_6_5, 0.5_6_1_8, 0.4_6, 0.6_2_3_1, 0.5_1_7_2]     )
        _UpperCAmelCase          =   1e-2 if torch_device != """mps""" else 3e-2

        assert np.abs(image_slice.flatten() - expected_slice     ).max() < tolerance
        assert np.abs(image_from_tuple_slice.flatten() - expected_slice     ).max() < tolerance







@slow
@require_torch
class 		a_						( unittest.TestCase	):







    def      _snake_case					(			self			:       List[Any]     )		->Optional[Any]:






        '''simple docstring'''






        _UpperCAmelCase          =   LDMPipeline.from_pretrained("""CompVis/ldm-celebahq-256"""     )
        ldm.to(__UpperCamelCase     )
        ldm.set_progress_bar_config(disable=__UpperCamelCase     )

        _UpperCAmelCase          =   torch.manual_seed(0     )
        _UpperCAmelCase          =   ldm(generator=__UpperCamelCase     ,					num_inference_steps=5     ,					output_type="""numpy"""     ).images

        _UpperCAmelCase          =   image[0, -3:, -3:, -1]

        assert image.shape == (1, 2_56, 2_56, 3)
        _UpperCAmelCase          =   np.array([0.4_3_9_9, 0.4_4_9_7_5, 0.4_6_8_2_5, 0.4_7_4, 0.4_3_5_9, 0.4_5_8_1, 0.4_5_0_9_5, 0.4_3_4_1, 0.4_4_4_7]     )
        _UpperCAmelCase          =   1e-2 if torch_device != """mps""" else 3e-2

        assert np.abs(image_slice.flatten() - expected_slice     ).max() < tolerance