

'''simple docstring'''



from typing import TYPE_CHECKING

from ....utils import _LazyModule


UpperCAmelCase :       Optional[Any]			   =     {'tokenization_tapex': ['TapexTokenizer']}


if TYPE_CHECKING:
  from .tokenization_tapex import TapexTokenizer


else:
  import sys

  UpperCAmelCase :       List[Any]			   =     _LazyModule(__name__, globals()['__file__'], _import_structure)









'''simple docstring'''



import logging
import os
import threading
import time


try:
  import warnings
except ImportError:
  UpperCAmelCase :       Optional[Any]			   =     None

try:
  import msvcrt
except ImportError:
  UpperCAmelCase :       List[Any]			   =     None

try:
  import fcntl
except ImportError:
  UpperCAmelCase :       int			   =     None


# Backward compatibility
# ------------------------------------------------
try:
  TimeoutError
except NameError:
  UpperCAmelCase :       Union[str, Any]			   =     OSError


# Data
# ------------------------------------------------
UpperCAmelCase :       List[Any]			   =     [
    'Timeout',
    'BaseFileLock',
    'WindowsFileLock',
    'UnixFileLock',
    'SoftFileLock',
    'FileLock',
]

UpperCAmelCase :       Tuple			   =     '3.0.12'


UpperCAmelCase :       str			   =     None







def    a__       (  ):




     """simple docstring"""


     global _logger
     __SCREAMING_SNAKE_CASE     =      _logger or logging.getLogger(__name__						)
     return _logger







class 						lowerCAmelCase__   (      a					):



 """simple docstring"""



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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      lock_file
      return None


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


      """simple docstring"""




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







class 						lowerCAmelCase__   :



 """simple docstring"""



 def __init__(		self					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       List[Any]			)  -> Union[str, Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      lock
      return None
 def __enter__(		self					:       List[str]			)  -> List[Any]:


      """simple docstring"""




      return self.lock


 def __exit__(		self					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       List[str]					,							__SCREAMING_SNAKE_CASE					:       Dict					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]			)  -> List[str]:


      """simple docstring"""




      self.lock.release()
      return None







class 						lowerCAmelCase__   :



 """simple docstring"""



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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      max_filename_length if max_filename_length is not None else 255
      # Hash the filename if it's too long
      __SCREAMING_SNAKE_CASE     =      self.hash_filename_if_too_long(__SCREAMING_SNAKE_CASE					,							__SCREAMING_SNAKE_CASE			)
      # The path to the lock file.
      __SCREAMING_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.
      __SCREAMING_SNAKE_CASE     =      None

      # The default timeout value.
      __SCREAMING_SNAKE_CASE     =      timeout

      # We use this lock primarily for the lock counter.
      __SCREAMING_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.
      __SCREAMING_SNAKE_CASE     =      0
      return None
 @property
 def       UpperCAmelCase__		(		self					:       List[Any]			)  -> List[str]:


      """simple docstring"""




      return self._lock_file
 @property
 def       UpperCAmelCase__		(		self					:       Optional[Any]			)  -> Optional[int]:


      """simple docstring"""




      return self._timeout
 @timeout.setter
 def       UpperCAmelCase__		(		self					:       List[str]					,							__SCREAMING_SNAKE_CASE					:       Dict			)  -> Optional[int]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      float(__SCREAMING_SNAKE_CASE			)
      return None
 def       UpperCAmelCase__		(		self					:       Optional[Any]			)  -> List[Any]:


      """simple docstring"""




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


      """simple docstring"""




      raise NotImplementedError()
 @property
 def       UpperCAmelCase__		(		self					:       Union[str, Any]			)  -> int:


      """simple docstring"""




      return self._lock_file_fd is not None
 def       UpperCAmelCase__		(		self					:       Optional[Any]					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]=None					,							__SCREAMING_SNAKE_CASE					:       Optional[int]=0.05			)  -> Optional[Any]:


      """simple docstring"""




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

      __SCREAMING_SNAKE_CASE     =      id(self			)
      __SCREAMING_SNAKE_CASE     =      self._lock_file
      __SCREAMING_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(__SCREAMING_SNAKE_CASE			)
      except:  # noqa
           # Something did go wrong, so decrement the counter.
           with self._thread_lock:
                __SCREAMING_SNAKE_CASE     =      max(0					,							self._lock_counter - 1			)

           raise
      return _Acquire_ReturnProxy(lock=self			)
 def       UpperCAmelCase__		(		self					:       Optional[Any]					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]=False			)  -> Dict:


      """simple docstring"""




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

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

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

      return None
 def __enter__(		self					:       Union[str, Any]			)  -> Optional[int]:


      """simple docstring"""




      self.acquire()
      return self
 def __exit__(		self					:       Optional[Any]					,							__SCREAMING_SNAKE_CASE					:       Tuple					,							__SCREAMING_SNAKE_CASE					:       Union[str, Any]					,							__SCREAMING_SNAKE_CASE					:       Any			)  -> Tuple:


      """simple docstring"""




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


      """simple docstring"""




      self.release(force=__SCREAMING_SNAKE_CASE			)
      return None


 def       UpperCAmelCase__		(		self					:       Optional[int]					,							__SCREAMING_SNAKE_CASE					:       str					,							__SCREAMING_SNAKE_CASE					:       int			)  -> str:


      """simple docstring"""




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







class 						lowerCAmelCase__   (      a					):



 """simple docstring"""



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


      """simple docstring"""




      from .file_utils import relative_to_absolute_path

      super().__init__(__SCREAMING_SNAKE_CASE					,							timeout=__SCREAMING_SNAKE_CASE					,							max_filename_length=__SCREAMING_SNAKE_CASE			)
      __SCREAMING_SNAKE_CASE     =      """\\\\?\\""" + relative_to_absolute_path(self.lock_file			)
 def       UpperCAmelCase__		(		self					:       Dict			)  -> Optional[int]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      os.O_RDWR | os.O_CREAT | os.O_TRUNC

      try:
           __SCREAMING_SNAKE_CASE     =      os.open(self._lock_file					,							__SCREAMING_SNAKE_CASE			)
      except OSError:
           pass
      else:
           try:
                msvcrt.locking(__SCREAMING_SNAKE_CASE					,							msvcrt.LK_NBLCK					,							1			)
           except OSError:
                os.close(__SCREAMING_SNAKE_CASE			)
           else:
                __SCREAMING_SNAKE_CASE     =      fd
      return None


 def       UpperCAmelCase__		(		self					:       Tuple			)  -> List[str]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      self._lock_file_fd
      __SCREAMING_SNAKE_CASE     =      None
      msvcrt.locking(__SCREAMING_SNAKE_CASE					,							msvcrt.LK_UNLCK					,							1			)
      os.close(__SCREAMING_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					:       Union[str, Any]					,							__SCREAMING_SNAKE_CASE					:       int					,							__SCREAMING_SNAKE_CASE					:       Union[str, Any]=-1					,							__SCREAMING_SNAKE_CASE					:       Optional[Any]=None			)  -> Any:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      os.statvfs(os.path.dirname(__SCREAMING_SNAKE_CASE			)			).f_namemax
      super().__init__(__SCREAMING_SNAKE_CASE					,							timeout=__SCREAMING_SNAKE_CASE					,							max_filename_length=__SCREAMING_SNAKE_CASE			)
 def       UpperCAmelCase__		(		self					:       Optional[int]			)  -> List[Any]:


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      os.O_RDWR | os.O_CREAT | os.O_TRUNC
      __SCREAMING_SNAKE_CASE     =      os.open(self._lock_file					,							__SCREAMING_SNAKE_CASE			)

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


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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      self._lock_file_fd
      __SCREAMING_SNAKE_CASE     =      None
      fcntl.flock(__SCREAMING_SNAKE_CASE					,							fcntl.LOCK_UN			)
      os.close(__SCREAMING_SNAKE_CASE			)
      return None







class 						lowerCAmelCase__   (      a					):



 """simple docstring"""



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


      """simple docstring"""




      __SCREAMING_SNAKE_CASE     =      os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
      try:
           __SCREAMING_SNAKE_CASE     =      os.open(self._lock_file					,							__SCREAMING_SNAKE_CASE			)
      except OSError:
           pass
      else:
           __SCREAMING_SNAKE_CASE     =      fd
      return None


 def       UpperCAmelCase__		(		self					:       int			)  -> Optional[int]:


      """simple docstring"""




      os.close(self._lock_file_fd			)
      __SCREAMING_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


UpperCAmelCase :       Dict			   =     None

if msvcrt:
  UpperCAmelCase :       Optional[int]			   =     WindowsFileLock
elif fcntl:
  UpperCAmelCase :       Optional[Any]			   =     UnixFileLock
else:
  UpperCAmelCase :       int			   =     SoftFileLock

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






