






'''simple docstring'''



import logging
import os
import threading
import time


try:
 import warnings
except ImportError:
 __UpperCamelCase      :       Any      			=      None

try:
 import msvcrt
except ImportError:
 __UpperCamelCase      :       str      			=      None

try:
 import fcntl
except ImportError:
 __UpperCamelCase      :       Tuple      			=      None


# Backward compatibility
# ------------------------------------------------
try:
 TimeoutError
except NameError:
 __UpperCamelCase      :       List[str]      			=      OSError


# Data
# ------------------------------------------------
__UpperCamelCase      :       Tuple      			=      [
    "Timeout",
    "BaseFileLock",
    "WindowsFileLock",
    "UnixFileLock",
    "SoftFileLock",
    "FileLock",
]

__UpperCamelCase      :       Optional[int]      			=      "3.0.12"


__UpperCamelCase      :       List[Any]      			=      None



def    __UpperCAmelCase	(				)  ->			Dict:

   """simple docstring"""
   global _logger
   __a				     =    _logger or logging.getLogger(__name__    )
   return _logger
class 							__SCREAMING_SNAKE_CASE       (							_UpperCAmelCase     ):




      def __init__(  self ,   lowerCamelCase      )							->Optional[int]:



         '''simple docstring'''
         __a				     =    lock_file
         return None






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



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







class 							__SCREAMING_SNAKE_CASE       :




      def __init__(  self ,   lowerCamelCase      )							->Dict:



         '''simple docstring'''
         __a				     =    lock
         return None




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



         '''simple docstring'''
         return self.lock






      def __exit__(  self ,   lowerCamelCase ,   lowerCamelCase ,   lowerCamelCase      )							->str:



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







class 							__SCREAMING_SNAKE_CASE       :




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



         '''simple docstring'''
         __a				     =    max_filename_length if max_filename_length is not None else 255
         # Hash the filename if it's too long
         __a				     =    self.hash_filename_if_too_long(A_ ,   A_      )
         # 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      )							->Tuple:



         '''simple docstring'''
         return self._lock_file




      @property
      def  __UpperCamelCase							(  self      )							->Any:



         '''simple docstring'''
         return self._timeout




      @timeout.setter
      def  __UpperCamelCase							(  self ,   lowerCamelCase      )							->Union[str, Any]:



         '''simple docstring'''
         __a				     =    float(A_      )
         return None




      def  __UpperCamelCase							(  self      )							->Tuple:



         '''simple docstring'''
         raise NotImplementedError()




      def  __UpperCamelCase							(  self      )							->Optional[Any]:



         '''simple docstring'''
         raise NotImplementedError()




      @property
      def  __UpperCamelCase							(  self      )							->List[str]:



         '''simple docstring'''
         return self._lock_file_fd is not None




      def  __UpperCamelCase							(  self ,   lowerCamelCase=None ,   lowerCamelCase=0.05      )							->Any:



         '''simple docstring'''
         # 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(A_      )
         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      )							->int:



         '''simple docstring'''
         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      )							->Optional[int]:



         '''simple docstring'''
         self.acquire()
         return self




      def __exit__(  self ,   lowerCamelCase ,   lowerCamelCase ,   lowerCamelCase      )							->List[Any]:



         '''simple docstring'''
         self.release()
         return None




      def __del__(  self      )							->Any:



         '''simple docstring'''
         self.release(force=A_      )
         return None






      def  __UpperCamelCase							(  self ,   lowerCamelCase ,   lowerCamelCase      )							->str:



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







class 							__SCREAMING_SNAKE_CASE       (							_UpperCAmelCase     ):




      def __init__(  self ,   lowerCamelCase ,   lowerCamelCase=-1 ,   lowerCamelCase=None      )							->Tuple:



         '''simple docstring'''
         from .file_utils import relative_to_absolute_path

         super().__init__(A_ ,   timeout=A_ ,   max_filename_length=A_      )
         __a				     =    '\\\\?\\' + relative_to_absolute_path(self.lock_file      )




      def  __UpperCamelCase							(  self      )							->List[str]:



         '''simple docstring'''
         __a				     =    os.O_RDWR | os.O_CREAT | os.O_TRUNC

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






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



         '''simple docstring'''
         __a				     =    self._lock_file_fd
         __a				     =    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 							__SCREAMING_SNAKE_CASE       (							_UpperCAmelCase     ):




      def __init__(  self ,   lowerCamelCase ,   lowerCamelCase=-1 ,   lowerCamelCase=None      )							->str:



         '''simple docstring'''
         __a				     =    os.statvfs(os.path.dirname(A_      )      ).f_namemax
         super().__init__(A_ ,   timeout=A_ ,   max_filename_length=A_      )




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



         '''simple docstring'''
         __a				     =    os.O_RDWR | os.O_CREAT | os.O_TRUNC
         __a				     =    os.open(self._lock_file ,   A_      )

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






      def  __UpperCamelCase							(  self      )							->Optional[Any]:



         '''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
         __a				     =    self._lock_file_fd
         __a				     =    None
         fcntl.flock(A_ ,   fcntl.LOCK_UN      )
         os.close(A_      )
         return None







class 							__SCREAMING_SNAKE_CASE       (							_UpperCAmelCase     ):




      def  __UpperCamelCase							(  self      )							->int:



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






      def  __UpperCamelCase							(  self      )							->Dict:



         '''simple docstring'''
         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      :       str      			=      None

if msvcrt:
 __UpperCamelCase      :       Union[str, Any]      			=      WindowsFileLock
elif fcntl:
 __UpperCamelCase      :       Optional[int]      			=      UnixFileLock
else:
 __UpperCamelCase      :       Dict      			=      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_torch_available


__UpperCamelCase      :       Tuple      			=      {
    """configuration_blip_2""": [
        """BLIP_2_PRETRAINED_CONFIG_ARCHIVE_MAP""",
        """Blip2Config""",
        """Blip2QFormerConfig""",
        """Blip2VisionConfig""",
    ],
    """processing_blip_2""": ["""Blip2Processor"""],
}

try:
 if not is_torch_available():
  raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
 pass
else:
 __UpperCamelCase      :       Optional[int]      			=      [
     """BLIP_2_PRETRAINED_MODEL_ARCHIVE_LIST""",
     """Blip2Model""",
     """Blip2QFormerModel""",
     """Blip2PreTrainedModel""",
     """Blip2ForConditionalGeneration""",
     """Blip2VisionModel""",
 ]

if TYPE_CHECKING:
 from .configuration_blip_a import (
     BLIP_2_PRETRAINED_CONFIG_ARCHIVE_MAP,
     BlipaConfig,
     BlipaQFormerConfig,
     BlipaVisionConfig,
 )
 from .processing_blip_a import BlipaProcessor

 try:
  if not is_torch_available():
   raise OptionalDependencyNotAvailable()
 except OptionalDependencyNotAvailable:
  pass
 else:
  from .modeling_blip_a import (
      BLIP_2_PRETRAINED_MODEL_ARCHIVE_LIST,
      BlipaForConditionalGeneration,
      BlipaModel,
      BlipaPreTrainedModel,
      BlipaQFormerModel,
      BlipaVisionModel,
  )

else:
 import sys

 __UpperCamelCase      :       Dict      			=      _LazyModule(__name__, globals()["""__file__"""], _import_structure, module_spec=__spec__)