

"""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 		lowerCamelCase__		(       ):

    '''simple docstring'''



    global _logger
    _lowerCAmelCase							: List[Any]	=     _logger or logging.getLogger(__name__ )
    return _logger







class 						__UpperCamelCase     (     __lowercase    ):

  def __init__(		self     ,_A   ):
      '''simple docstring'''

      _lowerCAmelCase							: Optional[int]	=     lock_file
      return None







  def __str__(		self   ):
      '''simple docstring'''

      _lowerCAmelCase							: Tuple	=     F"""The file lock \'{self.lock_file}\' could not be acquired."""
      return temp







class 						__UpperCamelCase     :

  def __init__(		self     ,_A   ):
      '''simple docstring'''

      _lowerCAmelCase							: Tuple	=     lock
      return None

  def __enter__(		self   ):
      '''simple docstring'''

      return self.lock







  def __exit__(		self     ,_A     ,_A     ,_A   ):
      '''simple docstring'''

      self.lock.release()
      return None







class 						__UpperCamelCase     :

  def __init__(		self     ,_A     ,_A=-1     ,_A=None   ):
      '''simple docstring'''

      _lowerCAmelCase							: Tuple	=     max_filename_length if max_filename_length is not None else 255
      # Hash the filename if it's too long
      _lowerCAmelCase							: Optional[Any]	=     self.hash_filename_if_too_long(_A     ,_A   )
      # The path to the lock file.
      _lowerCAmelCase							: Union[str, 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							: Tuple	=     None

      # The default timeout value.
      _lowerCAmelCase							: Optional[int]	=     timeout

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

  @property
  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      return self._lock_file

  @property
  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      return self._timeout

  @timeout.setter
  def __lowerCamelCase		(		self     ,_A   ):
      '''simple docstring'''

      _lowerCAmelCase							: str	=     float(_A   )
      return None

  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      raise NotImplementedError()

  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      raise NotImplementedError()

  @property
  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      return self._lock_file_fd is not None

  def __lowerCamelCase		(		self     ,_A=None     ,_A=0.0_5   ):
      '''simple docstring'''

      if timeout is None:
          _lowerCAmelCase							: List[str]	=     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							: Optional[int]	=     id(self   )
      _lowerCAmelCase							: Any	=     self._lock_file
      _lowerCAmelCase							: List[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   ):
      '''simple docstring'''

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

              if self._lock_counter == 0 or force:
                  _lowerCAmelCase							: List[Any]	=     id(self   )
                  _lowerCAmelCase							: Union[str, Any]	=     self._lock_file

                  logger().debug(F"""Attempting to release lock {lock_id} on {lock_filename}"""   )
                  self._release()
                  _lowerCAmelCase							: Union[str, Any]	=     0
                  logger().debug(F"""Lock {lock_id} released on {lock_filename}"""   )

      return None

  def __enter__(		self   ):
      '''simple docstring'''

      self.acquire()
      return self

  def __exit__(		self     ,_A     ,_A     ,_A   ):
      '''simple docstring'''

      self.release()
      return None

  def __del__(		self   ):
      '''simple docstring'''

      self.release(force=_A   )
      return None







  def __lowerCamelCase		(		self     ,_A     ,_A   ):
      '''simple docstring'''

      _lowerCAmelCase							: List[str]	=     os.path.basename(_A   )
      if len(_A   ) > max_length and max_length > 0:
          _lowerCAmelCase							: Union[str, Any]	=     os.path.dirname(_A   )
          _lowerCAmelCase							: Optional[Any]	=     str(hash(_A   )   )
          _lowerCAmelCase							: Any	=     filename[: max_length - len(_A   ) - 8] + '...' + hashed_filename + '.lock'
          return os.path.join(_A     ,_A   )
      else:
          return path







class 						__UpperCamelCase     (     __lowercase    ):

  def __init__(		self     ,_A     ,_A=-1     ,_A=None   ):
      '''simple docstring'''

      from .file_utils import relative_to_absolute_path

      super().__init__(_A     ,timeout=_A     ,max_filename_length=_A   )
      _lowerCAmelCase							: Any	=     '\\\\?\\' + relative_to_absolute_path(self.lock_file   )

  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      _lowerCAmelCase							: Tuple	=     os.O_RDWR | os.O_CREAT | os.O_TRUNC

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







  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      _lowerCAmelCase							: Dict	=     self._lock_file_fd
      _lowerCAmelCase							: Union[str, 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 						__UpperCamelCase     (     __lowercase    ):

  def __init__(		self     ,_A     ,_A=-1     ,_A=None   ):
      '''simple docstring'''

      _lowerCAmelCase							: List[str]	=     os.statvfs(os.path.dirname(_A   )   ).f_namemax
      super().__init__(_A     ,timeout=_A     ,max_filename_length=_A   )

  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      _lowerCAmelCase							: str	=     os.O_RDWR | os.O_CREAT | os.O_TRUNC
      _lowerCAmelCase							: Dict	=     os.open(self._lock_file     ,_A   )

      try:
          fcntl.flock(_A     ,fcntl.LOCK_EX | fcntl.LOCK_NB   )
      except OSError:
          os.close(_A   )
      else:
          _lowerCAmelCase							: Optional[Any]	=     fd
      return None







  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      _lowerCAmelCase							: List[str]	=     self._lock_file_fd
      _lowerCAmelCase							: Tuple	=     None
      fcntl.flock(_A     ,fcntl.LOCK_UN   )
      os.close(_A   )
      return None







class 						__UpperCamelCase     (     __lowercase    ):

  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      _lowerCAmelCase							: List[str]	=     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							: int	=     fd
      return None







  def __lowerCamelCase		(		self   ):
      '''simple docstring'''

      os.close(self._lock_file_fd   )
      _lowerCAmelCase							: 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""")







from __future__ import annotations

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

lowerCamelCase__						=				TypeVar("T")
lowerCamelCase__						=				TypeVar("U")







class 							lowerCAmelCase__   ( Generic[T, U]						):

						def __init__(      self							,				a							,				a					)       ->   Optional[Any]:


								'''simple docstring'''

								_UpperCamelCase         =						key
								_UpperCamelCase         =						val
								_UpperCamelCase         =						None
								_UpperCamelCase         =						None







						def __repr__(      self					)       ->   str:


								'''simple docstring'''

								return (
								    F'Node: key: {self.key}, val: {self.val}, '
								    F'has next: {bool(self.next					)}, has prev: {bool(self.prev					)}'
								)







class 							lowerCAmelCase__   ( Generic[T, U]						):

						def __init__(      self					)       ->   None:


								'''simple docstring'''

								_UpperCamelCase         =						DoubleLinkedListNode(a							,				a					)
								_UpperCamelCase         =						DoubleLinkedListNode(a							,				a					)
								_UpperCamelCase     ,						_UpperCamelCase         =						self.rear, self.head

						def __repr__(      self					)       ->   str:


								'''simple docstring'''

								_UpperCamelCase         =						["""DoubleLinkedList"""]
								_UpperCamelCase         =						self.head
								while node.next is not None:
										rep.append(str(a					)					)
										_UpperCamelCase         =						node.next
								rep.append(str(self.rear					)					)
								return ",\n    ".join(a					)

						def 			A_					(      self							,				a					)       ->   None:


								'''simple docstring'''

								_UpperCamelCase         =						self.rear.prev

								# All nodes other than self.head are guaranteed to have non-None previous
								assert previous is not None

								_UpperCamelCase         =						node
								_UpperCamelCase         =						previous
								_UpperCamelCase         =						node
								_UpperCamelCase         =						self.rear







						def 			A_					(      self							,				a					)       ->   DoubleLinkedListNode[T, U] | None:


								'''simple docstring'''

								if node.prev is None or node.next is None:
										return None

								_UpperCamelCase         =						node.next
								_UpperCamelCase         =						node.prev
								_UpperCamelCase         =						None
								_UpperCamelCase         =						None
								return node







class 							lowerCAmelCase__   ( Generic[T, U]						):
						UpperCamelCase_				:			dict[Callable[[T], U], LRUCache[T, U]]						=		{}

						def __init__(      self							,				a					)       ->   int:


								'''simple docstring'''

								_UpperCamelCase         =						DoubleLinkedList()
								_UpperCamelCase         =						capacity
								_UpperCamelCase         =						0
								_UpperCamelCase         =						0
								_UpperCamelCase         =						0
								_UpperCamelCase         =						{}

						def __repr__(      self					)       ->   str:


								'''simple docstring'''

								return (
								    F'CacheInfo(hits={self.hits}, misses={self.miss}, '
								    F'capacity={self.capacity}, current size={self.num_keys})'
								)

						def __contains__(      self							,				a					)       ->   bool:


								'''simple docstring'''

								return key in self.cache

						def 			A_					(      self							,				a					)       ->   U | None:


								'''simple docstring'''

								if key in self.cache:
										self.hits += 1
										_UpperCamelCase         =						self.cache[key]
										_UpperCamelCase         =						self.list.remove(self.cache[key]					)
										assert node == value_node

										# node is guaranteed not None because it is in self.cache
										assert node is not None
										self.list.add(a					)
										return node.val
								self.miss += 1
								return None

						def 			A_					(      self							,				a							,				a					)       ->   None:


								'''simple docstring'''

								if key not in self.cache:
										if self.num_keys >= self.capacity:
												# delete first node (oldest) when over capacity
												_UpperCamelCase         =						self.list.head.next

												# guaranteed to have a non-None first node when num_keys > 0
												# explain to type checker via assertions
												assert first_node is not None
												assert first_node.key is not None
												assert (
												    self.list.remove(a					) is not None
												)  # node guaranteed to be in list assert node.key is not None

												del self.cache[first_node.key]
												self.num_keys -= 1
										_UpperCamelCase         =						DoubleLinkedListNode(a							,				a					)
										self.list.add(self.cache[key]					)
										self.num_keys += 1

								else:
										# bump node to the end of the list, update value
										_UpperCamelCase         =						self.list.remove(self.cache[key]					)
										assert node is not None  # node guaranteed to be in list
										_UpperCamelCase         =						value
										self.list.add(a					)







						@classmethod
						def 			A_					(      cls							,				a = 1_28					)       ->   Callable[[Callable[[T], U]], Callable[..., U]]:


								'''simple docstring'''

								def cache_decorator_inner(a					) -> Callable[..., U]:
										def cache_decorator_wrapper(*a					) -> U:
												if func not in cls.decorator_function_to_instance_map:
														_UpperCamelCase         =						LRUCache(a					)

												_UpperCamelCase         =						cls.decorator_function_to_instance_map[func].get(args[0]					)
												if result is None:
														_UpperCamelCase         =						func(*a					)
														cls.decorator_function_to_instance_map[func].put(args[0]							,				a					)
												return result

										def cache_info() -> LRUCache[T, U]:
												return cls.decorator_function_to_instance_map[func]

										setattr(a							,				"""cache_info"""							,				a					)  # noqa: B010

										return cache_decorator_wrapper

								return cache_decorator_inner


if __name__ == "__main__":
		import doctest

		doctest.testmod()



