

"""simple docstring"""


import logging
import os
import threading
import time


try:
  import warnings
except ImportError:
  UpperCAmelCase													=None

try:
  import msvcrt
except ImportError:
  UpperCAmelCase													=None

try:
  import fcntl
except ImportError:
  UpperCAmelCase													=None


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


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

UpperCAmelCase													="3.0.12"


UpperCAmelCase													=None



def   _A					(	):



      """simple docstring"""






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





class lowerCamelCase__   (				SCREAMING_SNAKE_CASE    ):



    '''simple docstring'''




    def __init__( self ,lowerCamelCase_    )    ->      Optional[Any]:
          A				  =		lock_file
          return None




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





class lowerCamelCase__   :



    '''simple docstring'''




    def __init__( self ,lowerCamelCase_    )    ->      List[Any]:
          A				  =		lock
          return None




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




    def __exit__( self ,lowerCamelCase_ ,lowerCamelCase_ ,lowerCamelCase_    )    ->      Optional[Any]:
          self.lock.release()
          return None





class lowerCamelCase__   :



    '''simple docstring'''




    def __init__( self ,lowerCamelCase_ ,lowerCamelCase_=-1 ,lowerCamelCase_=None    )    ->      Dict:
          A				  =		max_filename_length if max_filename_length is not None else 2_5_5
          # Hash the filename if it's too long
          A				  =		self.hash_filename_if_too_long(lowerCamelCase_ ,lowerCamelCase_    )
          # The path to the lock file.
          A				  =		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				  =		None

          # The default timeout value.
          A				  =		timeout

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




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




    @property
    def UpperCamelCase__		( self    )    ->      List[Any]:
          return self._timeout




    @timeout.setter
    def UpperCamelCase__		( self ,lowerCamelCase_    )    ->      Tuple:
          A				  =		float(lowerCamelCase_    )
          return None




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




    def UpperCamelCase__		( self    )    ->      int:
          raise NotImplementedError()




    @property
    def UpperCamelCase__		( self    )    ->      Union[str, Any]:
          return self._lock_file_fd is not None




    def UpperCamelCase__		( self ,lowerCamelCase_=None ,lowerCamelCase_=0.05    )    ->      Any:
          # Use the default timeout, if no timeout is provided.
          if timeout is None:
                A				  =		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				  =		id(self    )
          A				  =		self._lock_file
          A				  =		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:
                      A				  =		max(0 ,self._lock_counter - 1    )

                raise
          return _Acquire_ReturnProxy(lock=self    )




    def UpperCamelCase__		( self ,lowerCamelCase_=False    )    ->      Union[str, Any]:
          with self._thread_lock:
                if self.is_locked:
                      self._lock_counter -= 1

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

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

          return None




    def __enter__( self    )    ->      str:
          self.acquire()
          return self




    def __exit__( self ,lowerCamelCase_ ,lowerCamelCase_ ,lowerCamelCase_    )    ->      Dict:
          self.release()
          return None




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




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





class lowerCamelCase__   (				SCREAMING_SNAKE_CASE    ):



    '''simple docstring'''




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

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




    def UpperCamelCase__		( self    )    ->      Tuple:
          A				  =		os.O_RDWR | os.O_CREAT | os.O_TRUNC

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




    def UpperCamelCase__		( self    )    ->      Optional[Any]:
          A				  =		self._lock_file_fd
          A				  =		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 lowerCamelCase__   (				SCREAMING_SNAKE_CASE    ):



    '''simple docstring'''




    def __init__( self ,lowerCamelCase_ ,lowerCamelCase_=-1 ,lowerCamelCase_=None    )    ->      Tuple:
          A				  =		os.statvfs(os.path.dirname(lowerCamelCase_    )    ).f_namemax
          super().__init__(lowerCamelCase_ ,timeout=lowerCamelCase_ ,max_filename_length=lowerCamelCase_    )




    def UpperCamelCase__		( self    )    ->      str:
          A				  =		os.O_RDWR | os.O_CREAT | os.O_TRUNC
          A				  =		os.open(self._lock_file ,lowerCamelCase_    )

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




    def UpperCamelCase__		( self    )    ->      Tuple:
          # 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				  =		self._lock_file_fd
          A				  =		None
          fcntl.flock(lowerCamelCase_ ,fcntl.LOCK_UN    )
          os.close(lowerCamelCase_    )
          return None





class lowerCamelCase__   (				SCREAMING_SNAKE_CASE    ):



    '''simple docstring'''




    def UpperCamelCase__		( self    )    ->      str:
          A				  =		os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
          try:
                A				  =		os.open(self._lock_file ,lowerCamelCase_    )
          except OSError:
                pass
          else:
                A				  =		fd
          return None




    def UpperCamelCase__		( self    )    ->      Optional[Any]:
          os.close(self._lock_file_fd    )
          A				  =		None

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




UpperCAmelCase													=None

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

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







"""simple docstring"""


from typing import TYPE_CHECKING

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


UpperCAmelCase													={
    "configuration_informer": [
        "INFORMER_PRETRAINED_CONFIG_ARCHIVE_MAP",
        "InformerConfig",
    ],
}

try:
  if not is_torch_available():
    raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
  pass
else:
  UpperCAmelCase													=[
      "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

  UpperCAmelCase													=_LazyModule(__name__, globals()["__file__"], _import_structure, module_spec=__spec__)




