





'''simple docstring'''






import logging
import os
import threading
import time


try:
					import warnings
except ImportError:
					lowerCAmelCase       		=      None

try:
					import msvcrt
except ImportError:
					lowerCAmelCase       		=      None

try:
					import fcntl
except ImportError:
					lowerCAmelCase       		=      None


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


# Data
# ------------------------------------------------
lowerCAmelCase       		=      [
    """Timeout""",
    """BaseFileLock""",
    """WindowsFileLock""",
    """UnixFileLock""",
    """SoftFileLock""",
    """FileLock""",
]

lowerCAmelCase       		=      """3.0.12"""


lowerCAmelCase       		=      None






def      __A						(							):
		global _logger
		lowerCAmelCase     :    Dict						=   _logger or logging.getLogger(__name__	)
		return _logger
class 				lowerCamelCase  (      _A       ):



	def __init__(   self		,      a_	):
			lowerCAmelCase     :    Any						=   lock_file
			return None





	def __str__(   self	):
			lowerCAmelCase     :    str						=   F'''The file lock \'{self.lock_file}\' could not be acquired.'''
			return temp






class 				lowerCamelCase  :



	def __init__(   self		,      a_	):
			lowerCAmelCase     :    Union[str, Any]						=   lock
			return None



	def __enter__(   self	):
			return self.lock





	def __exit__(   self		,      a_		,      a_		,      a_	):
			self.lock.release()
			return None






class 				lowerCamelCase  :



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

			# The default timeout value.
			lowerCAmelCase     :    List[Any]						=   timeout

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



	@property
	def   _lowerCamelCase    (   self	):
			return self._lock_file



	@property
	def   _lowerCamelCase    (   self	):
			return self._timeout



	@timeout.setter
	def   _lowerCamelCase    (   self		,      a_	):
			lowerCAmelCase     :    str						=   float(a_	)
			return None



	def   _lowerCamelCase    (   self	):
			raise NotImplementedError()



	def   _lowerCamelCase    (   self	):
			raise NotImplementedError()



	@property
	def   _lowerCamelCase    (   self	):
			return self._lock_file_fd is not None



	def   _lowerCamelCase    (   self		,      a_=None		,      a_=0.05	):
			# Use the default timeout, if no timeout is provided.
			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     :    Any						=   id(self	)
			lowerCAmelCase     :    Dict						=   self._lock_file
			lowerCAmelCase     :    Union[str, Any]						=   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:
							lowerCAmelCase     :    Optional[Any]						=   max(0		,      self._lock_counter - 1	)

					raise
			return _Acquire_ReturnProxy(lock=self	)



	def   _lowerCamelCase    (   self		,      a_=False	):
			with self._thread_lock:
					if self.is_locked:
							self._lock_counter -= 1

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

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

			return None



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



	def __exit__(   self		,      a_		,      a_		,      a_	):
			self.release()
			return None



	def __del__(   self	):
			self.release(force=a_	)
			return None





	def   _lowerCamelCase    (   self		,      a_		,      a_	):
			lowerCAmelCase     :    List[Any]						=   os.path.basename(a_	)
			if len(a_	) > max_length and max_length > 0:
					lowerCAmelCase     :    Union[str, Any]						=   os.path.dirname(a_	)
					lowerCAmelCase     :    str						=   str(hash(a_	)	)
					lowerCAmelCase     :    Dict						=   filename[: max_length - len(a_	) - 8] + "..." + hashed_filename + ".lock"
					return os.path.join(a_		,      a_	)
			else:
					return path






class 				lowerCamelCase  (      _A       ):



	def __init__(   self		,      a_		,      a_=-1		,      a_=None	):
			from .file_utils import relative_to_absolute_path

			super().__init__(a_		,      timeout=a_		,      max_filename_length=a_	)
			lowerCAmelCase     :    List[Any]						=   "\\\\?\\" + relative_to_absolute_path(self.lock_file	)



	def   _lowerCamelCase    (   self	):
			lowerCAmelCase     :    Union[str, Any]						=   os.O_RDWR | os.O_CREAT | os.O_TRUNC

			try:
					lowerCAmelCase     :    Union[str, 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:
							lowerCAmelCase     :    Optional[int]						=   fd
			return None





	def   _lowerCamelCase    (   self	):
			lowerCAmelCase     :    Dict						=   self._lock_file_fd
			lowerCAmelCase     :    Dict						=   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 				lowerCamelCase  (      _A       ):



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



	def   _lowerCamelCase    (   self	):
			lowerCAmelCase     :    Tuple						=   os.O_RDWR | os.O_CREAT | os.O_TRUNC
			lowerCAmelCase     :    Optional[Any]						=   os.open(self._lock_file		,      a_	)

			try:
					fcntl.flock(a_		,      fcntl.LOCK_EX | fcntl.LOCK_NB	)
			except OSError:
					os.close(a_	)
			else:
					lowerCAmelCase     :    Union[str, Any]						=   fd
			return None





	def   _lowerCamelCase    (   self	):
			# 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     :    Dict						=   self._lock_file_fd
			lowerCAmelCase     :    Any						=   None
			fcntl.flock(a_		,      fcntl.LOCK_UN	)
			os.close(a_	)
			return None






class 				lowerCamelCase  (      _A       ):



	def   _lowerCamelCase    (   self	):
			lowerCAmelCase     :    Union[str, Any]						=   os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
			try:
					lowerCAmelCase     :    Tuple						=   os.open(self._lock_file		,      a_	)
			except OSError:
					pass
			else:
					lowerCAmelCase     :    Optional[int]						=   fd
			return None





	def   _lowerCamelCase    (   self	):
			os.close(self._lock_file_fd	)
			lowerCAmelCase     :    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






lowerCAmelCase       		=      None

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

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










'''simple docstring'''






def      __A						(							a_						:  int	):
		assert (
		    isinstance(a_      ,a_	) and number_of_steps > 0
		), f'''number_of_steps needs to be positive integer, your input {number_of_steps}'''
		if number_of_steps == 1:
				return 1
		lowerCAmelCase	,		lowerCAmelCase     :    int						=   1, 1
		for _ in range(number_of_steps - 1	):
				lowerCAmelCase	,		lowerCAmelCase     :    Union[str, Any]						=   current + previous, current
		return current


if __name__ == "__main__":
					import doctest

					doctest.testmod()



