




import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      __magic_name__  :  str					=       None

try:
      import msvcrt
except ImportError:
      __magic_name__  :  Dict					=       None

try:
      import fcntl
except ImportError:
      __magic_name__  :  Any					=       None


# Backward compatibility
# ------------------------------------------------
try:
      TimeoutError
except NameError:
      __magic_name__  :  List[Any]					=       OSError


# Data
# ------------------------------------------------
__magic_name__  :  Union[str, Any]					=       [
    """Timeout""",
    """BaseFileLock""",
    """WindowsFileLock""",
    """UnixFileLock""",
    """SoftFileLock""",
    """FileLock""",
]

__magic_name__  :  Dict					=       """3.0.12"""


__magic_name__  :  Optional[Any]					=       None
def 		a_							( ):
  global _logger
  lowerCAmelCase__										=      _logger or logging.getLogger(__name__				)
  return _logger





class     SCREAMING_SNAKE_CASE__		(_a			):
     def __init__(						self	:		Tuple       ,						__lowerCamelCase	:		int						):



       """simple docstring"""
       lowerCAmelCase__										=      lock_file
       return None



     def __str__(						self	:		int						):



       """simple docstring"""
       lowerCAmelCase__										=      F"""The file lock '{self.lock_file}' could not be acquired."""
       return temp





class     SCREAMING_SNAKE_CASE__		:
     def __init__(						self	:		Union[str, Any]       ,						__lowerCamelCase	:		int						):



       """simple docstring"""
       lowerCAmelCase__										=      lock
       return None
     def __enter__(						self	:		Union[str, Any]						):



       """simple docstring"""
       return self.lock



     def __exit__(						self	:		Optional[Any]       ,						__lowerCamelCase	:		Dict       ,						__lowerCamelCase	:		List[str]       ,						__lowerCamelCase	:		Optional[int]						):



       """simple docstring"""
       self.lock.release()
       return None





class     SCREAMING_SNAKE_CASE__		:
     def __init__(						self	:		List[str]       ,						__lowerCamelCase	:		Dict       ,						__lowerCamelCase	:		Any=-1       ,						__lowerCamelCase	:		Any=None						):



       """simple docstring"""
       lowerCAmelCase__										=      max_filename_length if max_filename_length is not None else 2_55
       # Hash the filename if it's too long
       lowerCAmelCase__										=      self.hash_filename_if_too_long(__lowerCamelCase       ,						__lowerCamelCase						)
       # The path to the lock file.
       lowerCAmelCase__										=      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__										=      None

       # The default timeout value.
       lowerCAmelCase__										=      timeout

       # We use this lock primarily for the lock counter.
       lowerCAmelCase__										=      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__										=      0
       return None
     @property
     def        A__      (						self	:		Optional[int]						):



       """simple docstring"""
       return self._lock_file
     @property
     def        A__      (						self	:		Dict						):



       """simple docstring"""
       return self._timeout
     @timeout.setter
     def        A__      (						self	:		List[str]       ,						__lowerCamelCase	:		Dict						):



       """simple docstring"""
       lowerCAmelCase__										=      float(__lowerCamelCase						)
       return None
     def        A__      (						self	:		Union[str, Any]						):



       """simple docstring"""
       raise NotImplementedError()
     def        A__      (						self	:		str						):



       """simple docstring"""
       raise NotImplementedError()
     @property
     def        A__      (						self	:		Tuple						):



       """simple docstring"""
       return self._lock_file_fd is not None
     def        A__      (						self	:		List[Any]       ,						__lowerCamelCase	:		Tuple=None       ,						__lowerCamelCase	:		Optional[int]=0.05						):



       """simple docstring"""
       # Use the default timeout, if no timeout is provided.
       if timeout is None:
         lowerCAmelCase__										=      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__										=      id(self						)
       lowerCAmelCase__										=      self._lock_file
       lowerCAmelCase__										=      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(__lowerCamelCase						)
       except:  # noqa
         # Something did go wrong, so decrement the counter.
         with self._thread_lock:
           lowerCAmelCase__										=      max(0       ,						self._lock_counter - 1						)

         raise
       return _Acquire_ReturnProxy(lock=self						)
     def        A__      (						self	:		Optional[int]       ,						__lowerCamelCase	:		Union[str, Any]=False						):



       """simple docstring"""
       with self._thread_lock:
         if self.is_locked:
           self._lock_counter -= 1

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

             logger().debug(F"""Attempting to release lock {lock_id} on {lock_filename}"""						)
             self._release()
             lowerCAmelCase__										=      0
             logger().debug(F"""Lock {lock_id} released on {lock_filename}"""						)

       return None
     def __enter__(						self	:		Dict						):



       """simple docstring"""
       self.acquire()
       return self
     def __exit__(						self	:		Any       ,						__lowerCamelCase	:		Dict       ,						__lowerCamelCase	:		Any       ,						__lowerCamelCase	:		Union[str, Any]						):



       """simple docstring"""
       self.release()
       return None
     def __del__(						self	:		Optional[Any]						):



       """simple docstring"""
       self.release(force=__lowerCamelCase						)
       return None



     def        A__      (						self	:		List[Any]       ,						__lowerCamelCase	:		str       ,						__lowerCamelCase	:		int						):



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





class     SCREAMING_SNAKE_CASE__		(_a			):
     def __init__(						self	:		List[Any]       ,						__lowerCamelCase	:		Optional[Any]       ,						__lowerCamelCase	:		Tuple=-1       ,						__lowerCamelCase	:		Tuple=None						):



       """simple docstring"""
       from .file_utils import relative_to_absolute_path

       super().__init__(__lowerCamelCase       ,						timeout=__lowerCamelCase       ,						max_filename_length=__lowerCamelCase						)
       lowerCAmelCase__										=      '''\\\\?\\''' + relative_to_absolute_path(self.lock_file						)
     def        A__      (						self	:		Dict						):



       """simple docstring"""
       lowerCAmelCase__										=      os.O_RDWR | os.O_CREAT | os.O_TRUNC

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



     def        A__      (						self	:		List[Any]						):



       """simple docstring"""
       lowerCAmelCase__										=      self._lock_file_fd
       lowerCAmelCase__										=      None
       msvcrt.locking(__lowerCamelCase       ,						msvcrt.LK_UNLCK       ,						1						)
       os.close(__lowerCamelCase						)

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





class     SCREAMING_SNAKE_CASE__		(_a			):
     def __init__(						self	:		Tuple       ,						__lowerCamelCase	:		Optional[int]       ,						__lowerCamelCase	:		List[str]=-1       ,						__lowerCamelCase	:		Optional[Any]=None						):



       """simple docstring"""
       lowerCAmelCase__										=      os.statvfs(os.path.dirname(__lowerCamelCase						)						).f_namemax
       super().__init__(__lowerCamelCase       ,						timeout=__lowerCamelCase       ,						max_filename_length=__lowerCamelCase						)
     def        A__      (						self	:		Union[str, Any]						):



       """simple docstring"""
       lowerCAmelCase__										=      os.O_RDWR | os.O_CREAT | os.O_TRUNC
       lowerCAmelCase__										=      os.open(self._lock_file       ,						__lowerCamelCase						)

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



     def        A__      (						self	:		List[str]						):



       """simple docstring"""
       # 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__										=      self._lock_file_fd
       lowerCAmelCase__										=      None
       fcntl.flock(__lowerCamelCase       ,						fcntl.LOCK_UN						)
       os.close(__lowerCamelCase						)
       return None





class     SCREAMING_SNAKE_CASE__		(_a			):
     def        A__      (						self	:		List[Any]						):



       """simple docstring"""
       lowerCAmelCase__										=      os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
       try:
         lowerCAmelCase__										=      os.open(self._lock_file       ,						__lowerCamelCase						)
       except OSError:
         pass
       else:
         lowerCAmelCase__										=      fd
       return None



     def        A__      (						self	:		Optional[Any]						):



       """simple docstring"""
       os.close(self._lock_file_fd						)
       lowerCAmelCase__										=      None

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



__magic_name__  :  Dict					=       None

if msvcrt:
      __magic_name__  :  List[Any]					=       WindowsFileLock
elif fcntl:
      __magic_name__  :  Dict					=       UnixFileLock
else:
      __magic_name__  :  Dict					=       SoftFileLock

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







from typing import TYPE_CHECKING

# rely on isort to merge the imports
from ...utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available


__magic_name__  :  Optional[Any]					=       {
    """configuration_informer""": [
        """INFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """InformerConfig""",
    ],
}

try:
      if not is_torch_available():
            raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
      pass
else:
      __magic_name__  :  int					=       [
          """INFORMER_PRETRAINED_MODEL_ARCHIVE_LIST""",
          """InformerForPrediction""",
          """InformerModel""",
          """InformerPreTrainedModel""",
      ]


if TYPE_CHECKING:
      from .configuration_informer import INFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP, InformerConfig

      try:
            if not is_torch_available():
                  raise OptionalDependencyNotAvailable()
      except OptionalDependencyNotAvailable:
            pass
      else:
            from .modeling_informer import (
                INFORMER_PRETRAINED_MODEL_ARCHIVE_LIST,
                InformerForPrediction,
                InformerModel,
                InformerPreTrainedModel,
            )

else:
      import sys

      __magic_name__  :  Any					=       _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)

