






"""simple docstring"""




import logging
import os
import threading
import time


try:
 import warnings
except ImportError:
 _snake_case          =   None

try:
 import msvcrt
except ImportError:
 _snake_case          =   None

try:
 import fcntl
except ImportError:
 _snake_case          =   None


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


# Data
# ------------------------------------------------
_snake_case          =   [
    'Timeout',
    'BaseFileLock',
    'WindowsFileLock',
    'UnixFileLock',
    'SoftFileLock',
    'FileLock',
]

_snake_case          =   '3.0.12'


_snake_case          =   None

def 			lowerCAmelCase__     (     ):



  '''simple docstring'''
  global _logger
  _a  :       List[str]             =	_logger or logging.getLogger(__name__				)
  return _logger






class        UpperCamelCase			(	snake_case_      ):


 def __init__(     self		:						int							,						UpperCAmelCase__		:						str							)					->       Optional[int]:
   _a  :       Optional[Any]             =	lock_file
   return None







 def __str__(     self		:						str							)					->       Dict:
   _a  :       Any             =	f"""The file lock '{self.lock_file}' could not be acquired."""
   return temp







class        UpperCamelCase			:


 def __init__(     self		:						Optional[int]							,						UpperCAmelCase__		:						Optional[int]							)					->       List[Any]:
   _a  :       Dict             =	lock
   return None


 def __enter__(     self		:						List[Any]							)					->       int:
   return self.lock







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







class        UpperCamelCase			:


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

   # The default timeout value.
   _a  :       Tuple             =	timeout

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


 @property
 def _lowercase	(     self		:						List[str]							)					->       Union[str, Any]:
   return self._lock_file


 @property
 def _lowercase	(     self		:						Optional[Any]							)					->       Dict:
   return self._timeout


 @timeout.setter
 def _lowercase	(     self		:						Union[str, Any]							,						UpperCAmelCase__		:						Tuple							)					->       Optional[Any]:
   _a  :       int             =	float(UpperCAmelCase__							)
   return None


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


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


 @property
 def _lowercase	(     self		:						Optional[Any]							)					->       Dict:
   return self._lock_file_fd is not None


 def _lowercase	(     self		:						List[Any]							,						UpperCAmelCase__		:						Optional[Any]=None							,						UpperCAmelCase__		:						Optional[Any]=0.0_5							)					->       Union[str, Any]:
   # Use the default timeout, if no timeout is provided.
   if timeout is None:
     _a  :       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

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

     raise
   return _Acquire_ReturnProxy(lock=self							)


 def _lowercase	(     self		:						Tuple							,						UpperCAmelCase__		:						Optional[Any]=False							)					->       List[Any]:
   with self._thread_lock:
     if self.is_locked:
       self._lock_counter -= 1

       if self._lock_counter == 0 or force:
         _a  :       str             =	id(self							)
         _a  :       Dict             =	self._lock_file

         logger().debug(f"""Attempting to release lock {lock_id} on {lock_filename}"""							)
         self._release()
         _a  :       int             =	0
         logger().debug(f"""Lock {lock_id} released on {lock_filename}"""							)

   return None


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


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


 def __del__(     self		:						Tuple							)					->       Optional[int]:
   self.release(force=UpperCAmelCase__							)
   return None







 def _lowercase	(     self		:						Tuple							,						UpperCAmelCase__		:						str							,						UpperCAmelCase__		:						int							)					->       str:
   _a  :       Union[str, Any]             =	os.path.basename(UpperCAmelCase__							)
   if len(UpperCAmelCase__							) > max_length and max_length > 0:
     _a  :       List[Any]             =	os.path.dirname(UpperCAmelCase__							)
     _a  :       Union[str, Any]             =	str(hash(UpperCAmelCase__							)							)
     _a  :       Tuple             =	filename[: max_length - len(UpperCAmelCase__							) - 8] + """...""" + hashed_filename + """.lock"""
     return os.path.join(UpperCAmelCase__							,						UpperCAmelCase__							)
   else:
     return path







class        UpperCamelCase			(	snake_case_      ):


 def __init__(     self		:						Any							,						UpperCAmelCase__		:						int							,						UpperCAmelCase__		:						Tuple=-1							,						UpperCAmelCase__		:						Any=None							)					->       Any:
   from .file_utils import relative_to_absolute_path

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


 def _lowercase	(     self		:						Optional[int]							)					->       List[Any]:
   _a  :       Tuple             =	os.O_RDWR | os.O_CREAT | os.O_TRUNC

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







 def _lowercase	(     self		:						List[Any]							)					->       int:
   _a  :       str             =	self._lock_file_fd
   _a  :       Any             =	None
   msvcrt.locking(UpperCAmelCase__							,						msvcrt.LK_UNLCK							,						1							)
   os.close(UpperCAmelCase__							)

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







class        UpperCamelCase			(	snake_case_      ):


 def __init__(     self		:						Tuple							,						UpperCAmelCase__		:						List[Any]							,						UpperCAmelCase__		:						List[str]=-1							,						UpperCAmelCase__		:						List[str]=None							)					->       int:
   _a  :       Dict             =	os.statvfs(os.path.dirname(UpperCAmelCase__							)							).f_namemax
   super().__init__(UpperCAmelCase__							,						timeout=UpperCAmelCase__							,						max_filename_length=UpperCAmelCase__							)


 def _lowercase	(     self		:						Tuple							)					->       Any:
   _a  :       List[str]             =	os.O_RDWR | os.O_CREAT | os.O_TRUNC
   _a  :       Optional[Any]             =	os.open(self._lock_file							,						UpperCAmelCase__							)

   try:
     fcntl.flock(UpperCAmelCase__							,						fcntl.LOCK_EX | fcntl.LOCK_NB							)
   except OSError:
     os.close(UpperCAmelCase__							)
   else:
     _a  :       Optional[int]             =	fd
   return None







 def _lowercase	(     self		:						List[Any]							)					->       Dict:
   # 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
   _a  :       List[str]             =	self._lock_file_fd
   _a  :       List[str]             =	None
   fcntl.flock(UpperCAmelCase__							,						fcntl.LOCK_UN							)
   os.close(UpperCAmelCase__							)
   return None







class        UpperCamelCase			(	snake_case_      ):


 def _lowercase	(     self		:						Tuple							)					->       Any:
   _a  :       Tuple             =	os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
   try:
     _a  :       Tuple             =	os.open(self._lock_file							,						UpperCAmelCase__							)
   except OSError:
     pass
   else:
     _a  :       Union[str, Any]             =	fd
   return None







 def _lowercase	(     self		:						Dict							)					->       int:
   os.close(self._lock_file_fd							)
   _a  :       List[Any]             =	None

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







_snake_case          =   None

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

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









"""simple docstring"""




from typing import TYPE_CHECKING

from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_tf_available, is_torch_available


_snake_case          =   {
    'configuration_transfo_xl': ['TRANSFO_XL_PRETRAINED_CONFIG_ARCHIVE_MAP', 'TransfoXLConfig'],
    'tokenization_transfo_xl': ['TransfoXLCorpus', 'TransfoXLTokenizer'],
}

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 _snake_case          =   [
     'TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST',
     'AdaptiveEmbedding',
     'TransfoXLForSequenceClassification',
     'TransfoXLLMHeadModel',
     'TransfoXLModel',
     'TransfoXLPreTrainedModel',
     'load_tf_weights_in_transfo_xl',
 ]

try:
 if not is_tf_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 _snake_case          =   [
     'TF_TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST',
     'TFAdaptiveEmbedding',
     'TFTransfoXLForSequenceClassification',
     'TFTransfoXLLMHeadModel',
     'TFTransfoXLMainLayer',
     'TFTransfoXLModel',
     'TFTransfoXLPreTrainedModel',
 ]


if TYPE_CHECKING:
 from .configuration_transfo_xl import TRANSFO_XL_PRETRAINED_CONFIG_ARCHIVE_MAP, TransfoXLConfig
 from .tokenization_transfo_xl import TransfoXLCorpus, TransfoXLTokenizer

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_transfo_xl import (
      TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST,
      AdaptiveEmbedding,
      TransfoXLForSequenceClassification,
      TransfoXLLMHeadModel,
      TransfoXLModel,
      TransfoXLPreTrainedModel,
      load_tf_weights_in_transfo_xl,
  )

 try:
  if not is_tf_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_tf_transfo_xl import (
      TF_TRANSFO_XL_PRETRAINED_MODEL_ARCHIVE_LIST,
      TFAdaptiveEmbedding,
      TFTransfoXLForSequenceClassification,
      TFTransfoXLLMHeadModel,
      TFTransfoXLMainLayer,
      TFTransfoXLModel,
      TFTransfoXLPreTrainedModel,
  )

else:
 import sys

 _snake_case          =   _LazyModule(__name__, globals()['__file__'], _import_structure, module_spec=__spec__)

