






import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      __A    	=   None

try:
      import msvcrt
except ImportError:
      __A    	=   None

try:
      import fcntl
except ImportError:
      __A    	=   None


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


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

__A    	=   '''3.0.12'''


__A    	=   None




def        _SCREAMING_SNAKE_CASE  (				)    -> str:







   """simple docstring"""
   global _logger
   __snake_case				:      str     							= _logger or logging.getLogger(__name__  )
   return _logger




class a_    (    UpperCamelCase__							):







   def __init__(self					,   __a)			->      List[str]:
      """simple docstring"""






      __snake_case				:      int     							= lock_file
      return None







   def __str__(self)			->      str:
      """simple docstring"""






      __snake_case				:      List[Any]     							= F"""The file lock \'{self.lock_file}\' could not be acquired."""
      return temp




class a_    :







   def __init__(self					,   __a)			->      Tuple:
      """simple docstring"""






      __snake_case				:      Optional[int]     							= lock
      return None







   def __enter__(self)			->      List[Any]:
      """simple docstring"""






      return self.lock







   def __exit__(self					,   __a					,   __a					,   __a)			->      List[Any]:
      """simple docstring"""






      self.lock.release()
      return None




class a_    :







   def __init__(self					,   __a					,   __a=-1					,   __a=None)			->      List[Any]:
      """simple docstring"""






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

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

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







   @property
   def     SCREAMING_SNAKE_CASE__					(self)			->      Dict:
      """simple docstring"""






      return self._lock_file







   @property
   def     SCREAMING_SNAKE_CASE__					(self)			->      Optional[Any]:
      """simple docstring"""






      return self._timeout







   @timeout.setter
   def     SCREAMING_SNAKE_CASE__					(self					,   __a)			->      Optional[int]:
      """simple docstring"""






      __snake_case				:      Dict     							= float(_a)
      return None







   def     SCREAMING_SNAKE_CASE__					(self)			->      Optional[Any]:
      """simple docstring"""






      raise NotImplementedError()







   def     SCREAMING_SNAKE_CASE__					(self)			->      int:
      """simple docstring"""






      raise NotImplementedError()







   @property
   def     SCREAMING_SNAKE_CASE__					(self)			->      Optional[Any]:
      """simple docstring"""






      return self._lock_file_fd is not None







   def     SCREAMING_SNAKE_CASE__					(self					,   __a=None					,   __a=0.05)			->      Dict:
      """simple docstring"""






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

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

         raise
      return _Acquire_ReturnProxy(lock=self)







   def     SCREAMING_SNAKE_CASE__					(self					,   __a=False)			->      Optional[int]:
      """simple docstring"""






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

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

               logger().debug(F"""Attempting to release lock {lock_id} on {lock_filename}""")
               self._release()
               __snake_case				:      Tuple     							= 0
               logger().debug(F"""Lock {lock_id} released on {lock_filename}""")

      return None







   def __enter__(self)			->      Any:
      """simple docstring"""






      self.acquire()
      return self







   def __exit__(self					,   __a					,   __a					,   __a)			->      int:
      """simple docstring"""






      self.release()
      return None







   def __del__(self)			->      List[Any]:
      """simple docstring"""






      self.release(force=_a)
      return None







   def     SCREAMING_SNAKE_CASE__					(self					,   __a					,   __a)			->      str:
      """simple docstring"""






      __snake_case				:      Optional[Any]     							= os.path.basename(_a)
      if len(_a) > max_length and max_length > 0:
         __snake_case				:      Dict     							= os.path.dirname(_a)
         __snake_case				:      Any     							= str(hash(_a))
         __snake_case				:      Tuple     							= filename[: max_length - len(_a) - 8] + """...""" + hashed_filename + """.lock"""
         return os.path.join(_a					,   _a)
      else:
         return path




class a_    (    UpperCamelCase__							):







   def __init__(self					,   __a					,   __a=-1					,   __a=None)			->      Union[str, Any]:
      """simple docstring"""






      from .file_utils import relative_to_absolute_path

      super().__init__(_a					,   timeout=_a					,   max_filename_length=_a)
      __snake_case				:      Optional[Any]     							= """\\\\?\\""" + relative_to_absolute_path(self.lock_file)







   def     SCREAMING_SNAKE_CASE__					(self)			->      Union[str, Any]:
      """simple docstring"""






      __snake_case				:      List[str]     							= os.O_RDWR | os.O_CREAT | os.O_TRUNC

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







   def     SCREAMING_SNAKE_CASE__					(self)			->      int:
      """simple docstring"""






      __snake_case				:      str     							= self._lock_file_fd
      __snake_case				:      Any     							= None
      msvcrt.locking(_a					,   msvcrt.LK_UNLCK					,   1)
      os.close(_a)

      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					,   __a					,   __a=-1					,   __a=None)			->      Optional[int]:
      """simple docstring"""






      __snake_case				:      List[Any]     							= os.statvfs(os.path.dirname(_a)).f_namemax
      super().__init__(_a					,   timeout=_a					,   max_filename_length=_a)







   def     SCREAMING_SNAKE_CASE__					(self)			->      Any:
      """simple docstring"""






      __snake_case				:      Any     							= os.O_RDWR | os.O_CREAT | os.O_TRUNC
      __snake_case				:      Any     							= os.open(self._lock_file					,   _a)

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







   def     SCREAMING_SNAKE_CASE__					(self)			->      Dict:
      """simple docstring"""






      __snake_case				:      Any     							= self._lock_file_fd
      __snake_case				:      Dict     							= None
      fcntl.flock(_a					,   fcntl.LOCK_UN)
      os.close(_a)
      return None




class a_    (    UpperCamelCase__							):







   def     SCREAMING_SNAKE_CASE__					(self)			->      Tuple:
      """simple docstring"""






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







   def     SCREAMING_SNAKE_CASE__					(self)			->      List[Any]:
      """simple docstring"""






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

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


__A    	=   None

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

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







'''simple docstring'''






from functools import lru_cache




@lru_cache
def        _SCREAMING_SNAKE_CASE  (				A  :       int  )    -> int:







   """simple docstring"""
   if num < 0:
      raise ValueError('Number should not be negative.'  )

   return 1 if num in (0, 1) else num * factorial(num - 1  )


if __name__ == "__main__":
      import doctest

      doctest.testmod()