


import logging
import os
import threading
import time


try:
						import warnings
except ImportError:
						__lowerCAmelCase	: List[str]  				=None

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

try:
						import fcntl
except ImportError:
						__lowerCAmelCase	: str  				=None


# Backward compatibility
# ------------------------------------------------
try:
						TimeoutError
except NameError:
						__lowerCAmelCase	: int  				=OSError


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

__lowerCAmelCase	: Dict  				='3.0.12'


__lowerCAmelCase	: Dict  				=None


def        _UpperCamelCase							(      ):
						global _logger
						__SCREAMING_SNAKE_CASE						:     Dict					=		_logger or logging.getLogger(__name__      )
						return _logger
class        _lowercase      (			A__       ):






				'''simple docstring'''



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



				def __str__(       self			:Dict	)		->			Union[str, Any]:
										__SCREAMING_SNAKE_CASE						:     List[str]					=		f'''The file lock \'{self.lock_file}\' could not be acquired.'''
										return temp




class        _lowercase      :






				'''simple docstring'''



				def __init__(       self			:List[Any]			,			lowerCAmelCase__			:List[Any]	)		->			Optional[Any]:
										__SCREAMING_SNAKE_CASE						:     str					=		lock
										return None


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



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




class        _lowercase      :






				'''simple docstring'''



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

										# The default timeout value.
										__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		timeout

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


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


				@property
				def 		__magic_name__(       self			:Union[str, Any]	)		->			int:
										return self._timeout


				@timeout.setter
				def 		__magic_name__(       self			:Any			,			lowerCAmelCase__			:str	)		->			Any:
										__SCREAMING_SNAKE_CASE						:     int					=		float(lowerCAmelCase__	)
										return None


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


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


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


				def 		__magic_name__(       self			:str			,			lowerCAmelCase__			:Optional[Any]=None			,			lowerCAmelCase__			:int=0.05	)		->			int:
										# Use the default timeout, if no timeout is provided.
										if timeout is None:
																__SCREAMING_SNAKE_CASE						:     Tuple					=		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						:     Tuple					=		id(self	)
										__SCREAMING_SNAKE_CASE						:     Dict					=		self._lock_file
										__SCREAMING_SNAKE_CASE						:     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(lowerCAmelCase__	)
										except:  # noqa
																# Something did go wrong, so decrement the counter.
																with self._thread_lock:
																						__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		max(0			,			self._lock_counter - 1	)

																raise
										return _Acquire_ReturnProxy(lock=self	)


				def 		__magic_name__(       self			:Dict			,			lowerCAmelCase__			:Any=False	)		->			Tuple:
										with self._thread_lock:
																if self.is_locked:
																						self._lock_counter -= 1

																						if self._lock_counter == 0 or force:
																												__SCREAMING_SNAKE_CASE						:     Any					=		id(self	)
																												__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		self._lock_file

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

										return None


				def __enter__(       self			:Tuple	)		->			List[Any]:
										self.acquire()
										return self


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


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



				def 		__magic_name__(       self			:int			,			lowerCAmelCase__			:str			,			lowerCAmelCase__			:int	)		->			str:
										__SCREAMING_SNAKE_CASE						:     str					=		os.path.basename(lowerCAmelCase__	)
										if len(lowerCAmelCase__	) > max_length and max_length > 0:
																__SCREAMING_SNAKE_CASE						:     str					=		os.path.dirname(lowerCAmelCase__	)
																__SCREAMING_SNAKE_CASE						:     Optional[int]					=		str(hash(lowerCAmelCase__	)	)
																__SCREAMING_SNAKE_CASE						:     List[str]					=		filename[: max_length - len(lowerCAmelCase__	) - 8] + '''...''' + hashed_filename + '''.lock'''
																return os.path.join(lowerCAmelCase__			,			lowerCAmelCase__	)
										else:
																return path




class        _lowercase      (			A__       ):






				'''simple docstring'''



				def __init__(       self			:Tuple			,			lowerCAmelCase__			:Union[str, Any]			,			lowerCAmelCase__			:Union[str, Any]=-1			,			lowerCAmelCase__			:Optional[int]=None	)		->			List[Any]:
										from .file_utils import relative_to_absolute_path

										super().__init__(lowerCAmelCase__			,			timeout=lowerCAmelCase__			,			max_filename_length=lowerCAmelCase__	)
										__SCREAMING_SNAKE_CASE						:     Tuple					=		'''\\\\?\\''' + relative_to_absolute_path(self.lock_file	)


				def 		__magic_name__(       self			:Tuple	)		->			Optional[Any]:
										__SCREAMING_SNAKE_CASE						:     Tuple					=		os.O_RDWR | os.O_CREAT | os.O_TRUNC

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



				def 		__magic_name__(       self			:Tuple	)		->			Optional[int]:
										__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		self._lock_file_fd
										__SCREAMING_SNAKE_CASE						:     int					=		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        _lowercase      (			A__       ):






				'''simple docstring'''



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


				def 		__magic_name__(       self			:Optional[int]	)		->			Optional[Any]:
										__SCREAMING_SNAKE_CASE						:     str					=		os.O_RDWR | os.O_CREAT | os.O_TRUNC
										__SCREAMING_SNAKE_CASE						:     Union[str, Any]					=		os.open(self._lock_file			,			lowerCAmelCase__	)

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



				def 		__magic_name__(       self			:Tuple	)		->			Optional[int]:
										# 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
										__SCREAMING_SNAKE_CASE						:     int					=		self._lock_file_fd
										__SCREAMING_SNAKE_CASE						:     List[str]					=		None
										fcntl.flock(lowerCAmelCase__			,			fcntl.LOCK_UN	)
										os.close(lowerCAmelCase__	)
										return None




class        _lowercase      (			A__       ):






				'''simple docstring'''



				def 		__magic_name__(       self			:Tuple	)		->			int:
										__SCREAMING_SNAKE_CASE						:     List[str]					=		os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
										try:
																__SCREAMING_SNAKE_CASE						:     int					=		os.open(self._lock_file			,			lowerCAmelCase__	)
										except OSError:
																pass
										else:
																__SCREAMING_SNAKE_CASE						:     int					=		fd
										return None



				def 		__magic_name__(       self			:List[Any]	)		->			Optional[int]:
										os.close(self._lock_file_fd	)
										__SCREAMING_SNAKE_CASE						:     Optional[Any]					=		None

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




__lowerCAmelCase	: Optional[int]  				=None

if msvcrt:
						__lowerCAmelCase	: str  				=WindowsFileLock
elif fcntl:
						__lowerCAmelCase	: Union[str, Any]  				=UnixFileLock
else:
						__lowerCAmelCase	: Union[str, Any]  				=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


lowerCamelCase_								=		{
    "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:
  lowerCamelCase_								=		[
      "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

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