





'''simple docstring'''




import logging
import os
import threading
import time


try:
 import warnings
except ImportError:
 _SCREAMING_SNAKE_CASE							     =			None

try:
 import msvcrt
except ImportError:
 _SCREAMING_SNAKE_CASE							     =			None

try:
 import fcntl
except ImportError:
 _SCREAMING_SNAKE_CASE							     =			None


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


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

_SCREAMING_SNAKE_CASE							     =			"3.0.12"


_SCREAMING_SNAKE_CASE							     =			None





def 	__lowerCamelCase			(      )		->  int:
   global _logger
   snake_case 						=   _logger or logging.getLogger(__name__		)
   return _logger
class 			_lowerCAmelCase		(			A__	):






     """simple docstring"""






     def __init__(   self							:			List[Any]   ,       __snake_case							:			Optional[Any]						)->	Dict:
        snake_case 						=   lock_file
        return None



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




class 			_lowerCAmelCase		:






     """simple docstring"""






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



     def __enter__(   self							:			int						)->	str:
        return self.lock



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




class 			_lowerCAmelCase		:






     """simple docstring"""






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

        # The default timeout value.
        snake_case 						=   timeout

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



     @property
     def      lowerCAmelCase			(   self							:			List[str]						)->	Optional[int]:
        return self._lock_file



     @property
     def      lowerCAmelCase			(   self							:			str						)->	Any:
        return self._timeout



     @timeout.setter
     def      lowerCAmelCase			(   self							:			str   ,       __snake_case							:			Optional[int]						)->	Optional[Any]:
        snake_case 						=   float(__snake_case						)
        return None



     def      lowerCAmelCase			(   self							:			List[str]						)->	Optional[Any]:
        raise NotImplementedError()



     def      lowerCAmelCase			(   self							:			Union[str, Any]						)->	List[str]:
        raise NotImplementedError()



     @property
     def      lowerCAmelCase			(   self							:			Tuple						)->	str:
        return self._lock_file_fd is not None



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

           raise
        return _Acquire_ReturnProxy(lock=self						)



     def      lowerCAmelCase			(   self							:			Dict   ,       __snake_case							:			Dict=False						)->	int:
        with self._thread_lock:
           if self.is_locked:
              self._lock_counter -= 1

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

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

        return None



     def __enter__(   self							:			Optional[int]						)->	int:
        self.acquire()
        return self



     def __exit__(   self							:			List[Any]   ,       __snake_case							:			List[Any]   ,       __snake_case							:			Any   ,       __snake_case							:			Optional[Any]						)->	Tuple:
        self.release()
        return None



     def __del__(   self							:			int						)->	Union[str, Any]:
        self.release(force=__snake_case						)
        return None



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




class 			_lowerCAmelCase		(			A__	):






     """simple docstring"""






     def __init__(   self							:			List[Any]   ,       __snake_case							:			List[Any]   ,       __snake_case							:			Union[str, Any]=-1   ,       __snake_case							:			Any=None						)->	Any:
        from .file_utils import relative_to_absolute_path

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



     def      lowerCAmelCase			(   self							:			Optional[Any]						)->	List[Any]:
        snake_case 						=   os.O_RDWR | os.O_CREAT | os.O_TRUNC

        try:
           snake_case 						=   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:
              snake_case 						=   fd
        return None



     def      lowerCAmelCase			(   self							:			Optional[Any]						)->	Union[str, Any]:
        snake_case 						=   self._lock_file_fd
        snake_case 						=   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 			_lowerCAmelCase		(			A__	):






     """simple docstring"""






     def __init__(   self							:			List[str]   ,       __snake_case							:			List[str]   ,       __snake_case							:			Union[str, Any]=-1   ,       __snake_case							:			Tuple=None						)->	Tuple:
        snake_case 						=   os.statvfs(os.path.dirname(__snake_case						)						).f_namemax
        super().__init__(__snake_case   ,       timeout=__snake_case   ,       max_filename_length=__snake_case						)



     def      lowerCAmelCase			(   self							:			int						)->	Dict:
        snake_case 						=   os.O_RDWR | os.O_CREAT | os.O_TRUNC
        snake_case 						=   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:
           snake_case 						=   fd
        return None



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




class 			_lowerCAmelCase		(			A__	):






     """simple docstring"""






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



     def      lowerCAmelCase			(   self							:			Any						)->	int:
        os.close(self._lock_file_fd						)
        snake_case 						=   None

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




_SCREAMING_SNAKE_CASE							     =			None

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

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








'''simple docstring'''




from __future__ import annotations





def 	__lowerCamelCase			(      __lowerCAmelCase       :							Any      ,    __lowerCAmelCase       :							str      ,    __lowerCAmelCase       :							Any      ,    __lowerCAmelCase       :							int		)		->  Tuple:  # noqa: E741
   while r - l > 1:
      snake_case 						=   (l + r) // 2
      if v[m] >= key:
         snake_case 						=   m
      else:
         snake_case 						=   m  # noqa: E741
   return r





def 	__lowerCamelCase			(      __lowerCAmelCase       :							list[int]		)		->  int:
   if len(__lowerCAmelCase		) == 0:
      return 0

   snake_case 						=   [0] * len(__lowerCAmelCase		)
   snake_case 						=   1

   snake_case 						=   v[0]

   for i in range(1      ,    len(__lowerCAmelCase		)		):
      if v[i] < tail[0]:
         snake_case 						=   v[i]
      elif v[i] > tail[length - 1]:
         snake_case 						=   v[i]
         length += 1
      else:
         snake_case 						=   v[i]

   return length


if __name__ == "__main__":
 import doctest

 doctest.testmod()

