


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   UpperCamelCase			(      ):
						global _logger
						lowerCAmelCase_	:     Any          = _logger or logging.getLogger(__name__)
						return _logger

class __snake_case		( __UpperCamelCase ):





		"""simple docstring"""







		def __init__(							self    :   List[str]						,lowerCAmelCase__    :   List[str]      )      ->							Dict:




								'''simple docstring'''

								lowerCAmelCase_	:     List[Any]          = lock_file
								return None



		def __str__(							self    :   Optional[int]      )      ->							Dict:




								'''simple docstring'''

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






class __snake_case		:





		"""simple docstring"""







		def __init__(							self    :   Tuple						,lowerCAmelCase__    :   Optional[int]      )      ->							Tuple:




								'''simple docstring'''

								lowerCAmelCase_	:     Any          = lock
								return None



		def __enter__(							self    :   Union[str, Any]      )      ->							List[str]:




								'''simple docstring'''

								return self.lock



		def __exit__(							self    :   Dict						,lowerCAmelCase__    :   Any						,lowerCAmelCase__    :   Tuple						,lowerCAmelCase__    :   Dict      )      ->							List[Any]:




								'''simple docstring'''

								self.lock.release()
								return None






class __snake_case		:





		"""simple docstring"""







		def __init__(							self    :   Optional[int]						,lowerCAmelCase__    :   List[Any]						,lowerCAmelCase__    :   Dict=-1						,lowerCAmelCase__    :   Any=None      )      ->							Union[str, Any]:




								'''simple docstring'''

								lowerCAmelCase_	:     int          = max_filename_length if max_filename_length is not None else 2_55
								# Hash the filename if it's too long
								lowerCAmelCase_	:     Dict          = self.hash_filename_if_too_long(lowerCAmelCase__						,lowerCAmelCase__      )
								# The path to the lock file.
								lowerCAmelCase_	:     Optional[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_	:     int          = timeout

								# We use this lock primarily for the lock counter.
								lowerCAmelCase_	:     Any          = 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    :   int      )      ->							Tuple:




								'''simple docstring'''

								return self._lock_file



		@property
		def  UpperCAmelCase_							(							self    :   str      )      ->							Tuple:




								'''simple docstring'''

								return self._timeout



		@timeout.setter
		def  UpperCAmelCase_							(							self    :   Optional[Any]						,lowerCAmelCase__    :   List[str]      )      ->							int:




								'''simple docstring'''

								lowerCAmelCase_	:     Any          = float(lowerCAmelCase__      )
								return None



		def  UpperCAmelCase_							(							self    :   List[Any]      )      ->							Dict:




								'''simple docstring'''

								raise NotImplementedError()



		def  UpperCAmelCase_							(							self    :   str      )      ->							Optional[int]:




								'''simple docstring'''

								raise NotImplementedError()



		@property
		def  UpperCAmelCase_							(							self    :   Any      )      ->							List[Any]:




								'''simple docstring'''

								return self._lock_file_fd is not None



		def  UpperCAmelCase_							(							self    :   Union[str, Any]						,lowerCAmelCase__    :   Union[str, Any]=None						,lowerCAmelCase__    :   Optional[int]=0.05      )      ->							List[str]:




								'''simple docstring'''

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

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

														raise
								return _Acquire_ReturnProxy(lock=self      )



		def  UpperCAmelCase_							(							self    :   str						,lowerCAmelCase__    :   Tuple=False      )      ->							str:




								'''simple docstring'''

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

																				if self._lock_counter == 0 or force:
																										lowerCAmelCase_	:     Optional[int]          = id(self      )
																										lowerCAmelCase_	:     Union[str, Any]          = self._lock_file

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

								return None



		def __enter__(							self    :   List[str]      )      ->							Dict:




								'''simple docstring'''

								self.acquire()
								return self



		def __exit__(							self    :   Optional[Any]						,lowerCAmelCase__    :   Union[str, Any]						,lowerCAmelCase__    :   Union[str, Any]						,lowerCAmelCase__    :   Optional[Any]      )      ->							Dict:




								'''simple docstring'''

								self.release()
								return None



		def __del__(							self    :   Dict      )      ->							Any:




								'''simple docstring'''

								self.release(force=lowerCAmelCase__      )
								return None



		def  UpperCAmelCase_							(							self    :   Tuple						,lowerCAmelCase__    :   Union[str, Any]						,lowerCAmelCase__    :   Any      )      ->							Any:




								'''simple docstring'''

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






class __snake_case		( __UpperCamelCase ):





		"""simple docstring"""







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




								'''simple docstring'''

								from .file_utils import relative_to_absolute_path

								super().__init__(lowerCAmelCase__						,timeout=lowerCAmelCase__						,max_filename_length=lowerCAmelCase__      )
								lowerCAmelCase_	:     Dict          = "\\\\?\\" + relative_to_absolute_path(self.lock_file      )



		def  UpperCAmelCase_							(							self    :   Dict      )      ->							int:




								'''simple docstring'''

								lowerCAmelCase_	:     Any          = os.O_RDWR | os.O_CREAT | os.O_TRUNC

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



		def  UpperCAmelCase_							(							self    :   Any      )      ->							Tuple:




								'''simple docstring'''

								lowerCAmelCase_	:     Any          = self._lock_file_fd
								lowerCAmelCase_	:     Optional[Any]          = None
								msvcrt.locking(lowerCAmelCase__						,msvcrt.LK_UNLCK						,1      )
								os.close(lowerCAmelCase__      )

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






class __snake_case		( __UpperCamelCase ):





		"""simple docstring"""







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




								'''simple docstring'''

								lowerCAmelCase_	:     Any          = os.statvfs(os.path.dirname(lowerCAmelCase__      )      ).f_namemax
								super().__init__(lowerCAmelCase__						,timeout=lowerCAmelCase__						,max_filename_length=lowerCAmelCase__      )



		def  UpperCAmelCase_							(							self    :   Tuple      )      ->							Optional[int]:




								'''simple docstring'''

								lowerCAmelCase_	:     int          = os.O_RDWR | os.O_CREAT | os.O_TRUNC
								lowerCAmelCase_	:     str          = os.open(self._lock_file						,lowerCAmelCase__      )

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



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




								'''simple docstring'''

								lowerCAmelCase_	:     Any          = self._lock_file_fd
								lowerCAmelCase_	:     Optional[Any]          = None
								fcntl.flock(lowerCAmelCase__						,fcntl.LOCK_UN      )
								os.close(lowerCAmelCase__      )
								return None






class __snake_case		( __UpperCamelCase ):





		"""simple docstring"""







		def  UpperCAmelCase_							(							self    :   Optional[int]      )      ->							List[str]:




								'''simple docstring'''

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



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




								'''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






_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''')




from __future__ import annotations

from collections.abc import Iterator
from typing import Generic, TypeVar

__magic_name__  							=	TypeVar('''T''')




class        _SCREAMING_SNAKE_CASE	(   Generic[T]			):
      def __init__(      self		,						lowerCamelCase			):
          snake_case__    =					data
          snake_case__    =					None



      def __str__(      self			):
          return F"""{self.data}"""





class        _SCREAMING_SNAKE_CASE	(   Generic[T]			):
      def __init__(      self			):
          snake_case__    =					None
      def __iter__(      self			):
          snake_case__    =					self.top
          while node:
              yield node.data
              snake_case__    =					node.next
      def __str__(      self			):
          return "->".join([str(lowerCamelCase			) for item in self]			)
      def __len__(      self			):
          return len(tuple(iter(self			)			)			)
      def A_      (      self			):
          return self.top is None
      def A_      (      self		,						lowerCamelCase			):
          snake_case__    =					Node(lowerCamelCase			)
          if not self.is_empty():
              snake_case__    =					self.top
          snake_case__    =					node
      def A_      (      self			):
          if self.is_empty():
              raise IndexError("pop from empty stack"			)
          assert isinstance(self.top		,						lowerCamelCase			)
          snake_case__    =					self.top
          snake_case__    =					self.top.next
          return pop_node.data
      def A_      (      self			):
          if self.is_empty():
              raise IndexError("peek from empty stack"			)

          assert self.top is not None
          return self.top.data



      def A_      (      self			):
          snake_case__    =					None


if __name__ == "__main__":
     from doctest import testmod

     testmod()



