
"""simple docstring"""






import logging
import os
import threading
import time


try:
					import warnings
except ImportError:
					lowerCamelCase__ :       Optional[int]						     =			None

try:
					import msvcrt
except ImportError:
					lowerCamelCase__ :       Any						     =			None

try:
					import fcntl
except ImportError:
					lowerCamelCase__ :       List[Any]						     =			None


# Backward compatibility
# ------------------------------------------------
try:
					TimeoutError
except NameError:
					lowerCamelCase__ :       Optional[Any]						     =			OSError


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

lowerCamelCase__ :       Union[str, Any]						     =			"3.0.12"


lowerCamelCase__ :       str						     =			None





def __A						(   )->							Dict:




		'''simple docstring'''






		global _logger
		SCREAMING_SNAKE_CASE       :     Any				= _logger or logging.getLogger(__name__     )
		return _logger




class 				lowercase__(				_UpperCAmelCase   ):
	'''simple docstring'''

	def __init__(						self    :int		, lowerCamelCase_    :List[Any]					)							->			Tuple:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= lock_file
			return None







	def __str__(						self    :List[str]					)							->			Any:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= f"The file lock '{self.lock_file}' could not be acquired."
			return temp





class 				lowercase__:
	'''simple docstring'''

	def __init__(						self    :str		, lowerCamelCase_    :Union[str, Any]					)							->			List[Any]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     List[Any]				= lock
			return None





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







			'''simple docstring'''




			return self.lock







	def __exit__(						self    :Optional[Any]		, lowerCamelCase_    :Tuple		, lowerCamelCase_    :List[str]		, lowerCamelCase_    :Optional[Any]					)							->			Any:







			'''simple docstring'''




			self.lock.release()
			return None





class 				lowercase__:
	'''simple docstring'''

	def __init__(						self    :Any		, lowerCamelCase_    :Optional[Any]		, lowerCamelCase_    :Tuple=-1		, lowerCamelCase_    :str=None					)							->			Dict:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     List[Any]				= max_filename_length if max_filename_length is not None else 2_55
			# 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       :     List[Any]				= 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       :     List[Any]				= None

			# The default timeout value.
			SCREAMING_SNAKE_CASE       :     Dict				= timeout

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





	@property
	def   __lowerCAmelCase (						self    :str					)							->			Any:







			'''simple docstring'''




			return self._lock_file





	@property
	def   __lowerCAmelCase (						self    :Optional[Any]					)							->			Tuple:







			'''simple docstring'''




			return self._timeout





	@timeout.setter
	def   __lowerCAmelCase (						self    :Tuple		, lowerCamelCase_    :List[str]					)							->			Union[str, Any]:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Dict				= float(lowerCamelCase_					)
			return None





	def   __lowerCAmelCase (						self    :Optional[int]					)							->			Tuple:







			'''simple docstring'''




			raise NotImplementedError()





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







			'''simple docstring'''




			raise NotImplementedError()





	@property
	def   __lowerCAmelCase (						self    :Dict					)							->			Any:







			'''simple docstring'''




			return self._lock_file_fd is not None





	def   __lowerCAmelCase (						self    :Optional[Any]		, lowerCamelCase_    :Any=None		, lowerCamelCase_    :Optional[int]=0.0_5					)							->			Union[str, Any]:







			'''simple docstring'''




			if timeout is None:
					SCREAMING_SNAKE_CASE       :     Optional[Any]				= 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       :     Union[str, Any]				= 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       :     Tuple				= max(0		, self._lock_counter - 1					)

					raise
			return _Acquire_ReturnProxy(lock=self					)





	def   __lowerCAmelCase (						self    :int		, lowerCamelCase_    :Dict=False					)							->			Tuple:







			'''simple docstring'''




			with self._thread_lock:
					if self.is_locked:
							self._lock_counter -= 1

							if self._lock_counter == 0 or force:
									SCREAMING_SNAKE_CASE       :     int				= id(self					)
									SCREAMING_SNAKE_CASE       :     List[str]				= self._lock_file

									logger().debug(f"Attempting to release lock {lock_id} on {lock_filename}"					)
									self._release()
									SCREAMING_SNAKE_CASE       :     Optional[Any]				= 0
									logger().debug(f"Lock {lock_id} released on {lock_filename}"					)

			return None





	def __enter__(						self    :Any					)							->			Optional[int]:







			'''simple docstring'''




			self.acquire()
			return self





	def __exit__(						self    :Tuple		, lowerCamelCase_    :Optional[int]		, lowerCamelCase_    :List[str]		, lowerCamelCase_    :Union[str, Any]					)							->			List[str]:







			'''simple docstring'''




			self.release()
			return None





	def __del__(						self    :List[Any]					)							->			int:







			'''simple docstring'''




			self.release(force=lowerCamelCase_					)
			return None







	def   __lowerCAmelCase (						self    :List[Any]		, lowerCamelCase_    :str		, lowerCamelCase_    :int					)							->			str:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[Any]				= os.path.basename(lowerCamelCase_					)
			if len(lowerCamelCase_					) > max_length and max_length > 0:
					SCREAMING_SNAKE_CASE       :     Optional[Any]				= os.path.dirname(lowerCamelCase_					)
					SCREAMING_SNAKE_CASE       :     Dict				= str(hash(lowerCamelCase_					)					)
					SCREAMING_SNAKE_CASE       :     Any				= filename[: max_length - len(lowerCamelCase_					) - 8] + '''...''' + hashed_filename + '''.lock'''
					return os.path.join(lowerCamelCase_		, lowerCamelCase_					)
			else:
					return path





class 				lowercase__(				_UpperCAmelCase   ):
	'''simple docstring'''

	def __init__(						self    :int		, lowerCamelCase_    :List[Any]		, lowerCamelCase_    :Optional[int]=-1		, lowerCamelCase_    :List[Any]=None					)							->			Union[str, Any]:







			'''simple docstring'''




			from .file_utils import relative_to_absolute_path

			super().__init__(lowerCamelCase_		, timeout=lowerCamelCase_		, max_filename_length=lowerCamelCase_					)
			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= '''\\\\?\\''' + relative_to_absolute_path(self.lock_file					)





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







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Any				= os.O_RDWR | os.O_CREAT | os.O_TRUNC

			try:
					SCREAMING_SNAKE_CASE       :     int				= 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       :     Any				= fd
			return None







	def   __lowerCAmelCase (						self    :List[Any]					)							->			int:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     List[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__(				_UpperCAmelCase   ):
	'''simple docstring'''

	def __init__(						self    :List[str]		, lowerCamelCase_    :List[str]		, lowerCamelCase_    :Optional[Any]=-1		, lowerCamelCase_    :Any=None					)							->			int:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     str				= os.statvfs(os.path.dirname(lowerCamelCase_					)					).f_namemax
			super().__init__(lowerCamelCase_		, timeout=lowerCamelCase_		, max_filename_length=lowerCamelCase_					)





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







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Tuple				= os.O_RDWR | os.O_CREAT | os.O_TRUNC
			SCREAMING_SNAKE_CASE       :     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   __lowerCAmelCase (						self    :List[str]					)							->			Dict:







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[Any]				= self._lock_file_fd
			SCREAMING_SNAKE_CASE       :     Union[str, Any]				= None
			fcntl.flock(lowerCamelCase_		, fcntl.LOCK_UN					)
			os.close(lowerCamelCase_					)
			return None





class 				lowercase__(				_UpperCAmelCase   ):
	'''simple docstring'''

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







			'''simple docstring'''




			SCREAMING_SNAKE_CASE       :     Optional[Any]				= os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
			try:
					SCREAMING_SNAKE_CASE       :     List[str]				= os.open(self._lock_file		, lowerCamelCase_					)
			except OSError:
					pass
			else:
					SCREAMING_SNAKE_CASE       :     Optional[int]				= fd
			return None







	def   __lowerCAmelCase (						self    :List[str]					)							->			List[Any]:







			'''simple docstring'''




			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__ :       Union[str, Any]						     =			None

if msvcrt:
					lowerCamelCase__ :       Union[str, Any]						     =			WindowsFileLock
elif fcntl:
					lowerCamelCase__ :       Dict						     =			UnixFileLock
else:
					lowerCamelCase__ :       List[str]						     =			SoftFileLock

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







"""simple docstring"""






import os
import sys

lowerCamelCase__ :       List[Any]						     =			os.path.join(os.path.dirname(__file__), "src")
sys.path.append(SRC_DIR)


from transformers import (
    AutoConfig,
    AutoModel,
    AutoModelForCausalLM,
    AutoModelForMaskedLM,
    AutoModelForQuestionAnswering,
    AutoModelForSequenceClassification,
    AutoTokenizer,
    add_start_docstrings,
)


lowerCamelCase__ :       str						     =			[
    "torch",
    "numpy",
    "tokenizers",
    "filelock",
    "requests",
    "tqdm",
    "regex",
    "sentencepiece",
    "sacremoses",
    "importlib_metadata",
    "huggingface_hub",
]





@add_start_docstrings(AutoConfig.__doc__     )
def __A						(   *a_ :							Any		,		**a_ :							Union[str, Any]     )->							Dict:




		'''simple docstring'''






		return AutoConfig.from_pretrained(*a_		,		**a_     )





@add_start_docstrings(AutoTokenizer.__doc__     )
def __A						(   *a_ :							str		,		**a_ :							Union[str, Any]     )->							Union[str, Any]:




		'''simple docstring'''






		return AutoTokenizer.from_pretrained(*a_		,		**a_     )





@add_start_docstrings(AutoModel.__doc__     )
def __A						(   *a_ :							List[str]		,		**a_ :							int     )->							Dict:




		'''simple docstring'''






		return AutoModel.from_pretrained(*a_		,		**a_     )





@add_start_docstrings(AutoModelForCausalLM.__doc__     )
def __A						(   *a_ :							Any		,		**a_ :							Tuple     )->							Dict:




		'''simple docstring'''






		return AutoModelForCausalLM.from_pretrained(*a_		,		**a_     )





@add_start_docstrings(AutoModelForMaskedLM.__doc__     )
def __A						(   *a_ :							Dict		,		**a_ :							Optional[Any]     )->							Optional[int]:




		'''simple docstring'''






		return AutoModelForMaskedLM.from_pretrained(*a_		,		**a_     )





@add_start_docstrings(AutoModelForSequenceClassification.__doc__     )
def __A						(   *a_ :							Optional[int]		,		**a_ :							str     )->							Optional[int]:




		'''simple docstring'''






		return AutoModelForSequenceClassification.from_pretrained(*a_		,		**a_     )





@add_start_docstrings(AutoModelForQuestionAnswering.__doc__     )
def __A						(   *a_ :							List[str]		,		**a_ :							int     )->							List[Any]:




		'''simple docstring'''






		return AutoModelForQuestionAnswering.from_pretrained(*a_		,		**a_     )





