


import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      __UpperCAmelCase      :Union[str, Any]			    =       None

try:
      import msvcrt
except ImportError:
      __UpperCAmelCase      :Tuple			    =       None

try:
      import fcntl
except ImportError:
      __UpperCAmelCase      :str			    =       None


# Backward compatibility
# ------------------------------------------------
try:
      TimeoutError
except NameError:
      __UpperCAmelCase      :Dict			    =       OSError


# Data
# ------------------------------------------------
__UpperCAmelCase      :Any			    =       [
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

__UpperCAmelCase      :Tuple			    =       "3.0.12"


__UpperCAmelCase      :int			    =       None




def      _a     (						):





     '''simple docstring'''






     global _logger
     __UpperCAmelCase     :			Optional[int]  							=      _logger or logging.getLogger(__name__  )
     return _logger



class      a    (		_a	):


     """simple docstring"""






     def __init__(    self					:		str							,			snake_case					:		Optional[int]      )						->							Optional[Any]:
          __UpperCAmelCase     :			Any  							=      lock_file
          return None






     def __str__(    self					:		str      )						->							Union[str, Any]:
          __UpperCAmelCase     :			Tuple  							=      f'The file lock \'{self.lock_file}\' could not be acquired.'
          return temp



class      a    :


     """simple docstring"""






     def __init__(    self					:		Optional[Any]							,			snake_case					:		Dict      )						->							List[Any]:
          __UpperCAmelCase     :			Tuple  							=      lock
          return None

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






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



class      a    :


     """simple docstring"""






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

          # The default timeout value.
          __UpperCAmelCase     :			Optional[Any]  							=      timeout

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

     @property
     def 			lowerCamelCase__			(    self					:		Dict      )						->							Optional[Any]:
          return self._lock_file

     @property
     def 			lowerCamelCase__			(    self					:		str      )						->							Dict:
          return self._timeout

     @timeout.setter
     def 			lowerCamelCase__			(    self					:		Optional[Any]							,			snake_case					:		str      )						->							Any:
          __UpperCAmelCase     :			Union[str, Any]  							=      float(snake_case      )
          return None

     def 			lowerCamelCase__			(    self					:		Optional[Any]      )						->							Dict:
          raise NotImplementedError()

     def 			lowerCamelCase__			(    self					:		int      )						->							Any:
          raise NotImplementedError()

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

     def 			lowerCamelCase__			(    self					:		Optional[Any]							,			snake_case					:		Tuple=None							,			snake_case					:		List[Any]=0.05      )						->							int:
          # Use the default timeout, if no timeout is provided.
          if timeout is None:
               __UpperCAmelCase     :			Dict  							=      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     :			Optional[Any]  							=      id(self      )
          __UpperCAmelCase     :			Any  							=      self._lock_file
          __UpperCAmelCase     :			Dict  							=      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(snake_case      )
          except:  # noqa
               # Something did go wrong, so decrement the counter.
               with self._thread_lock:
                    __UpperCAmelCase     :			Dict  							=      max(0							,			self._lock_counter - 1      )

               raise
          return _Acquire_ReturnProxy(lock=self      )

     def 			lowerCamelCase__			(    self					:		Optional[int]							,			snake_case					:		Any=False      )						->							int:
          with self._thread_lock:
               if self.is_locked:
                    self._lock_counter -= 1

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

                         logger().debug(f'Attempting to release lock {lock_id} on {lock_filename}'      )
                         self._release()
                         __UpperCAmelCase     :			str  							=      0
                         logger().debug(f'Lock {lock_id} released on {lock_filename}'      )

          return None

     def __enter__(    self					:		List[Any]      )						->							Tuple:
          self.acquire()
          return self

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

     def __del__(    self					:		List[str]      )						->							List[str]:
          self.release(force=snake_case      )
          return None






     def 			lowerCamelCase__			(    self					:		Optional[Any]							,			snake_case					:		str							,			snake_case					:		int      )						->							str:
          __UpperCAmelCase     :			Any  							=      os.path.basename(snake_case      )
          if len(snake_case      ) > max_length and max_length > 0:
               __UpperCAmelCase     :			Optional[Any]  							=      os.path.dirname(snake_case      )
               __UpperCAmelCase     :			Tuple  							=      str(hash(snake_case      )      )
               __UpperCAmelCase     :			List[Any]  							=      filename[: max_length - len(snake_case      ) - 8] + '''...''' + hashed_filename + '''.lock'''
               return os.path.join(snake_case							,			snake_case      )
          else:
               return path



class      a    (		_a	):


     """simple docstring"""






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

          super().__init__(snake_case							,			timeout=snake_case							,			max_filename_length=snake_case      )
          __UpperCAmelCase     :			List[str]  							=      '''\\\\?\\''' + relative_to_absolute_path(self.lock_file      )

     def 			lowerCamelCase__			(    self					:		List[Any]      )						->							int:
          __UpperCAmelCase     :			Optional[int]  							=      os.O_RDWR | os.O_CREAT | os.O_TRUNC

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






     def 			lowerCamelCase__			(    self					:		str      )						->							str:
          __UpperCAmelCase     :			Dict  							=      self._lock_file_fd
          __UpperCAmelCase     :			Optional[Any]  							=      None
          msvcrt.locking(snake_case							,			msvcrt.LK_UNLCK							,			1      )
          os.close(snake_case      )

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



class      a    (		_a	):


     """simple docstring"""






     def __init__(    self					:		Optional[int]							,			snake_case					:		Optional[int]							,			snake_case					:		List[Any]=-1							,			snake_case					:		List[Any]=None      )						->							int:
          __UpperCAmelCase     :			Optional[Any]  							=      os.statvfs(os.path.dirname(snake_case      )      ).f_namemax
          super().__init__(snake_case							,			timeout=snake_case							,			max_filename_length=snake_case      )

     def 			lowerCamelCase__			(    self					:		Optional[int]      )						->							Dict:
          __UpperCAmelCase     :			Optional[int]  							=      os.O_RDWR | os.O_CREAT | os.O_TRUNC
          __UpperCAmelCase     :			Union[str, Any]  							=      os.open(self._lock_file							,			snake_case      )

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






     def 			lowerCamelCase__			(    self					:		int      )						->							List[str]:
          # 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
          __UpperCAmelCase     :			List[Any]  							=      self._lock_file_fd
          __UpperCAmelCase     :			str  							=      None
          fcntl.flock(snake_case							,			fcntl.LOCK_UN      )
          os.close(snake_case      )
          return None



class      a    (		_a	):


     """simple docstring"""






     def 			lowerCamelCase__			(    self					:		Tuple      )						->							Optional[Any]:
          __UpperCAmelCase     :			Union[str, Any]  							=      os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
          try:
               __UpperCAmelCase     :			List[Any]  							=      os.open(self._lock_file							,			snake_case      )
          except OSError:
               pass
          else:
               __UpperCAmelCase     :			str  							=      fd
          return None






     def 			lowerCamelCase__			(    self					:		Optional[Any]      )						->							Optional[int]:
          os.close(self._lock_file_fd      )
          __UpperCAmelCase     :			Optional[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      :Optional[Any]			    =       None

if msvcrt:
      __UpperCAmelCase      :Any			    =       WindowsFileLock
elif fcntl:
      __UpperCAmelCase      :List[Any]			    =       UnixFileLock
else:
      __UpperCAmelCase      :Dict			    =       SoftFileLock

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



'''simple docstring'''



import os
import sys
import unittest


__UpperCAmelCase      :Optional[int]			    =       os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
sys.path.append(os.path.join(git_repo_path, "utils"))

import check_dummies  # noqa: E402
from check_dummies import create_dummy_files, create_dummy_object, find_backend, read_init  # noqa: E402


# Align TRANSFORMERS_PATH in check_dummies with the current path
__UpperCAmelCase      :Dict			    =       os.path.join(git_repo_path, "src", "diffusers")



class      a    (		unittest.TestCase	):


     """simple docstring"""






     def 			lowerCamelCase__			(    self					:		Any      )						->							int:
          __UpperCAmelCase     :			Optional[Any]  							=      find_backend('''    if not is_torch_available():'''      )
          self.assertEqual(snake_case							,			'''torch'''      )

          # backend_with_underscore = find_backend("    if not is_tensorflow_text_available():")
          # self.assertEqual(backend_with_underscore, "tensorflow_text")

          __UpperCAmelCase     :			Union[str, Any]  							=      find_backend('''    if not (is_torch_available() and is_transformers_available()):'''      )
          self.assertEqual(snake_case							,			'''torch_and_transformers'''      )

          # double_backend_with_underscore = find_backend(
          #    "    if not (is_sentencepiece_available() and is_tensorflow_text_available()):"
          # )
          # self.assertEqual(double_backend_with_underscore, "sentencepiece_and_tensorflow_text")

          __UpperCAmelCase     :			List[str]  							=      find_backend(
              '''    if not (is_torch_available() and is_transformers_available() and is_onnx_available()):'''      )
          self.assertEqual(snake_case							,			'''torch_and_transformers_and_onnx'''      )

     def 			lowerCamelCase__			(    self					:		Optional[int]      )						->							int:
          __UpperCAmelCase     :			Tuple  							=      read_init()
          # We don't assert on the exact list of keys to allow for smooth grow of backend-specific objects
          self.assertIn('''torch'''							,			snake_case      )
          self.assertIn('''torch_and_transformers'''							,			snake_case      )
          self.assertIn('''flax_and_transformers'''							,			snake_case      )
          self.assertIn('''torch_and_transformers_and_onnx'''							,			snake_case      )

          # Likewise, we can't assert on the exact content of a key
          self.assertIn('''UNet2DModel'''							,			objects['''torch''']      )
          self.assertIn('''FlaxUNet2DConditionModel'''							,			objects['''flax''']      )
          self.assertIn('''StableDiffusionPipeline'''							,			objects['''torch_and_transformers''']      )
          self.assertIn('''FlaxStableDiffusionPipeline'''							,			objects['''flax_and_transformers''']      )
          self.assertIn('''LMSDiscreteScheduler'''							,			objects['''torch_and_scipy''']      )
          self.assertIn('''OnnxStableDiffusionPipeline'''							,			objects['''torch_and_transformers_and_onnx''']      )

     def 			lowerCamelCase__			(    self					:		Optional[int]      )						->							List[Any]:
          __UpperCAmelCase     :			str  							=      create_dummy_object('''CONSTANT'''							,			'''\'torch\''''      )
          self.assertEqual(snake_case							,			'''\nCONSTANT = None\n'''      )

          __UpperCAmelCase     :			Union[str, Any]  							=      create_dummy_object('''function'''							,			'''\'torch\''''      )
          self.assertEqual(
              snake_case							,			'''\ndef function(*args, **kwargs):\n    requires_backends(function, \'torch\')\n'''      )

          __UpperCAmelCase     :			Optional[int]  							=      '''
class FakeClass(metaclass=DummyObject):
    _backends = \'torch\'

    def __init__(self, *args, **kwargs):
        requires_backends(self, \'torch\')

    @classmethod
    def from_config(cls, *args, **kwargs):
        requires_backends(cls, \'torch\')

    @classmethod
    def from_pretrained(cls, *args, **kwargs):
        requires_backends(cls, \'torch\')
'''
          __UpperCAmelCase     :			Optional[Any]  							=      create_dummy_object('''FakeClass'''							,			'''\'torch\''''      )
          self.assertEqual(snake_case							,			snake_case      )






     def 			lowerCamelCase__			(    self					:		int      )						->							List[Any]:
          __UpperCAmelCase     :			List[str]  							=      '''# This file is autogenerated by the command `make fix-copies`, do not edit.
from ..utils import DummyObject, requires_backends


CONSTANT = None


def function(*args, **kwargs):
    requires_backends(function, ["torch"])


class FakeClass(metaclass=DummyObject):
    _backends = ["torch"]

    def __init__(self, *args, **kwargs):
        requires_backends(self, ["torch"])

    @classmethod
    def from_config(cls, *args, **kwargs):
        requires_backends(cls, ["torch"])

    @classmethod
    def from_pretrained(cls, *args, **kwargs):
        requires_backends(cls, ["torch"])
'''
          __UpperCAmelCase     :			Optional[int]  							=      create_dummy_files({'''torch''': ['''CONSTANT''', '''function''', '''FakeClass''']}      )
          self.assertEqual(dummy_files['''torch''']							,			snake_case      )