
import logging
import os
import threading
import time


try:
		import warnings
except ImportError:
		lowercase   =	None

try:
		import msvcrt
except ImportError:
		lowercase   =	None

try:
		import fcntl
except ImportError:
		lowercase   =	None


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


# Data
# ------------------------------------------------
lowercase   =	[
    'Timeout',
    'BaseFileLock',
    'WindowsFileLock',
    'UnixFileLock',
    'SoftFileLock',
    'FileLock',
]

lowercase   =	'3.0.12'


lowercase   =	None



def   __lowerCAmelCase    (				)	->     List[str]:
							global _logger
							lowerCamelCase_					=       _logger or logging.getLogger(__name__							)
							return _logger





class    __A(					_lowerCamelCase							):

					def __init__(						self       :  Tuple ,       __UpperCamelCase       :  Union[str, Any] ):
												lowerCamelCase_					=       lock_file
												return None



					def __str__(						self       :  List[Any] ):
												lowerCamelCase_					=       F'''The file lock \'{self.lock_file}\' could not be acquired.'''
												return temp





class    __A:

					def __init__(						self       :  int ,       __UpperCamelCase       :  Union[str, Any] ):
												lowerCamelCase_					=       lock
												return None

					def __enter__(						self       :  Dict ):
												return self.lock



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





class    __A:

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

												# The default timeout value.
												lowerCamelCase_					=       timeout

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

					@property
					def  lowercase__   (						self       :  str ):
												return self._lock_file

					@property
					def  lowercase__   (						self       :  Tuple ):
												return self._timeout

					@timeout.setter
					def  lowercase__   (						self       :  Union[str, Any] ,       __UpperCamelCase       :  Any ):
												lowerCamelCase_					=       float(__UpperCamelCase )
												return None

					def  lowercase__   (						self       :  Optional[Any] ):
												raise NotImplementedError()

					def  lowercase__   (						self       :  int ):
												raise NotImplementedError()

					@property
					def  lowercase__   (						self       :  List[Any] ):
												return self._lock_file_fd is not None

					def  lowercase__   (						self       :  int ,       __UpperCamelCase       :  List[str]=None ,       __UpperCamelCase       :  Any=0.05 ):
												# Use the default timeout, if no timeout is provided.
												if timeout is None:
																			lowerCamelCase_					=       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_					=       id(self )
												lowerCamelCase_					=       self._lock_file
												lowerCamelCase_					=       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:
																										lowerCamelCase_					=       max(0 ,       self._lock_counter - 1 )

																			raise
												return _Acquire_ReturnProxy(lock=self )

					def  lowercase__   (						self       :  Optional[int] ,       __UpperCamelCase       :  Tuple=False ):
												with self._thread_lock:
																			if self.is_locked:
																										self._lock_counter -= 1

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

																																	logger().debug(F'''Attempting to release lock {lock_id} on {lock_filename}''' )
																																	self._release()
																																	lowerCamelCase_					=       0
																																	logger().debug(F'''Lock {lock_id} released on {lock_filename}''' )

												return None

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

					def __exit__(						self       :  List[str] ,       __UpperCamelCase       :  Dict ,       __UpperCamelCase       :  Optional[Any] ,       __UpperCamelCase       :  str ):
												self.release()
												return None

					def __del__(						self       :  Union[str, Any] ):
												self.release(force=__UpperCamelCase )
												return None



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





class    __A(					_lowerCamelCase							):

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

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

					def  lowercase__   (						self       :  Any ):
												lowerCamelCase_					=       os.O_RDWR | os.O_CREAT | os.O_TRUNC

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



					def  lowercase__   (						self       :  Optional[Any] ):
												lowerCamelCase_					=       self._lock_file_fd
												lowerCamelCase_					=       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(					_lowerCamelCase							):

					def __init__(						self       :  List[Any] ,       __UpperCamelCase       :  Any ,       __UpperCamelCase       :  List[Any]=-1 ,       __UpperCamelCase       :  Dict=None ):
												lowerCamelCase_					=       os.statvfs(os.path.dirname(__UpperCamelCase ) ).f_namemax
												super().__init__(__UpperCamelCase ,       timeout=__UpperCamelCase ,       max_filename_length=__UpperCamelCase )

					def  lowercase__   (						self       :  Any ):
												lowerCamelCase_					=       os.O_RDWR | os.O_CREAT | os.O_TRUNC
												lowerCamelCase_					=       os.open(self._lock_file ,       __UpperCamelCase )

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



					def  lowercase__   (						self       :  Optional[int] ):
												# 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
												lowerCamelCase_					=       self._lock_file_fd
												lowerCamelCase_					=       None
												fcntl.flock(__UpperCamelCase ,       fcntl.LOCK_UN )
												os.close(__UpperCamelCase )
												return None





class    __A(					_lowerCamelCase							):

					def  lowercase__   (						self       :  int ):
												lowerCamelCase_					=       os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
												try:
																			lowerCamelCase_					=       os.open(self._lock_file ,       __UpperCamelCase )
												except OSError:
																			pass
												else:
																			lowerCamelCase_					=       fd
												return None



					def  lowercase__   (						self       :  Optional[Any] ):
												os.close(self._lock_file_fd )
												lowerCamelCase_					=       None

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

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

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










import unittest

from diffusers.pipelines.pipeline_utils import is_safetensors_compatible



class        lowerCAmelCase__					(  unittest.TestCase       ):


				def __UpperCamelCase						(							self      :  List[str]       )		->     Any:
											A      						=    [
											    'safety_checker/pytorch_model.bin',
											    'safety_checker/model.safetensors',
											    'vae/diffusion_pytorch_model.bin',
											    'vae/diffusion_pytorch_model.safetensors',
											    'text_encoder/pytorch_model.bin',
											    'text_encoder/model.safetensors',
											    'unet/diffusion_pytorch_model.bin',
											    'unet/diffusion_pytorch_model.safetensors',
											]
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  str       )		->     List[Any]:
											A      						=    [
											    'unet/diffusion_pytorch_model.bin',
											    'unet/diffusion_pytorch_model.safetensors',
											]
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Any       )		->     Tuple:
											A      						=    [
											    'safety_checker/pytorch_model.bin',
											    'safety_checker/model.safetensors',
											    'vae/diffusion_pytorch_model.bin',
											    'vae/diffusion_pytorch_model.safetensors',
											    'text_encoder/pytorch_model.bin',
											    'text_encoder/model.safetensors',
											    'unet/diffusion_pytorch_model.bin',
											    # Removed: 'unet/diffusion_pytorch_model.safetensors',
											]
											self.assertFalse(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Dict       )		->     Any:
											A      						=    [
											    'text_encoder/pytorch_model.bin',
											    'text_encoder/model.safetensors',
											]
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Optional[int]       )		->     List[str]:
											A      						=    [
											    'safety_checker/pytorch_model.bin',
											    'safety_checker/model.safetensors',
											    'vae/diffusion_pytorch_model.bin',
											    'vae/diffusion_pytorch_model.safetensors',
											    'text_encoder/pytorch_model.bin',
											    # Removed: 'text_encoder/model.safetensors',
											    'unet/diffusion_pytorch_model.bin',
											    'unet/diffusion_pytorch_model.safetensors',
											]
											self.assertFalse(is_safetensors_compatible(__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  List[str]       )		->     List[str]:
											A      						=    [
											    'safety_checker/pytorch_model.fp16.bin',
											    'safety_checker/model.fp16.safetensors',
											    'vae/diffusion_pytorch_model.fp16.bin',
											    'vae/diffusion_pytorch_model.fp16.safetensors',
											    'text_encoder/pytorch_model.fp16.bin',
											    'text_encoder/model.fp16.safetensors',
											    'unet/diffusion_pytorch_model.fp16.bin',
											    'unet/diffusion_pytorch_model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     Optional[Any]:
											A      						=    [
											    'unet/diffusion_pytorch_model.fp16.bin',
											    'unet/diffusion_pytorch_model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  int       )		->     Optional[int]:
											# pass variant but use the non-variant filenames
											A      						=    [
											    'unet/diffusion_pytorch_model.bin',
											    'unet/diffusion_pytorch_model.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Any       )		->     List[str]:
											A      						=    [
											    'safety_checker/pytorch_model.fp16.bin',
											    'safety_checker/model.fp16.safetensors',
											    'vae/diffusion_pytorch_model.fp16.bin',
											    'vae/diffusion_pytorch_model.fp16.safetensors',
											    'text_encoder/pytorch_model.fp16.bin',
											    'text_encoder/model.fp16.safetensors',
											    'unet/diffusion_pytorch_model.fp16.bin',
											    # Removed: 'unet/diffusion_pytorch_model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertFalse(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  Optional[Any]       )		->     Optional[int]:
											A      						=    [
											    'text_encoder/pytorch_model.fp16.bin',
											    'text_encoder/model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  List[str]       )		->     int:
											# pass variant but use the non-variant filenames
											A      						=    [
											    'text_encoder/pytorch_model.bin',
											    'text_encoder/model.safetensors',
											]
											A      						=    'fp16'
											self.assertTrue(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )


				def __UpperCamelCase						(							self      :  List[str]       )		->     Optional[Any]:
											A      						=    [
											    'safety_checker/pytorch_model.fp16.bin',
											    'safety_checker/model.fp16.safetensors',
											    'vae/diffusion_pytorch_model.fp16.bin',
											    'vae/diffusion_pytorch_model.fp16.safetensors',
											    'text_encoder/pytorch_model.fp16.bin',
											    # 'text_encoder/model.fp16.safetensors',
											    'unet/diffusion_pytorch_model.fp16.bin',
											    'unet/diffusion_pytorch_model.fp16.safetensors',
											]
											A      						=    'fp16'
											self.assertFalse(is_safetensors_compatible(__UpperCamelCase						, variant=__UpperCamelCase       )       )