"""simple docstring"""






import logging
import os
import threading
import time


try:
						import warnings
except ImportError:
						_A   					=	None

try:
						import msvcrt
except ImportError:
						_A   					=	None

try:
						import fcntl
except ImportError:
						_A   					=	None


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


# Data
# ------------------------------------------------
_A   					=	[
    '''Timeout''',
    '''BaseFileLock''',
    '''WindowsFileLock''',
    '''UnixFileLock''',
    '''SoftFileLock''',
    '''FileLock''',
]

_A   					=	'''3.0.12'''


_A   					=	None




def 						SCREAMING_SNAKE_CASE					(  )							->   Optional[int]:
				global _logger
				SCREAMING_SNAKE_CASE__		  =				_logger or logging.getLogger(__name__       )
				return _logger

class lowerCamelCase		(_UpperCAmelCase      ):






					'''simple docstring'''


					def __init__(						self			:				Any   , _snake_case			:				Dict		)							->       Tuple:
									SCREAMING_SNAKE_CASE__		  =				lock_file
									return None



					def __str__(						self			:				List[str]		)							->       Tuple:
									SCREAMING_SNAKE_CASE__		  =				F"""The file lock '{self.lock_file}' could not be acquired."""
									return temp



class lowerCamelCase		:






					'''simple docstring'''


					def __init__(						self			:				Tuple   , _snake_case			:				Any		)							->       Any:
									SCREAMING_SNAKE_CASE__		  =				lock
									return None



					def __enter__(						self			:				Any		)							->       str:
									return self.lock



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



class lowerCamelCase		:






					'''simple docstring'''


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

									# The default timeout value.
									SCREAMING_SNAKE_CASE__		  =				timeout

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



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



					@property
					def 			lowerCAmelCase_					(						self			:				Tuple		)							->       Any:
									return self._timeout



					@timeout.setter
					def 			lowerCAmelCase_					(						self			:				Union[str, Any]   , _snake_case			:				Any		)							->       int:
									SCREAMING_SNAKE_CASE__		  =				float(__UpperCamelCase		)
									return None



					def 			lowerCAmelCase_					(						self			:				Optional[int]		)							->       List[Any]:
									raise NotImplementedError()



					def 			lowerCAmelCase_					(						self			:				str		)							->       str:
									raise NotImplementedError()



					@property
					def 			lowerCAmelCase_					(						self			:				Optional[int]		)							->       int:
									return self._lock_file_fd is not None



					def 			lowerCAmelCase_					(						self			:				Any   , _snake_case			:				List[Any]=None   , _snake_case			:				int=0.05		)							->       Tuple:






									if timeout is None:
													SCREAMING_SNAKE_CASE__		  =				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__		  =				id(self		)
									SCREAMING_SNAKE_CASE__		  =				self._lock_file
									SCREAMING_SNAKE_CASE__		  =				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(__UpperCamelCase		)
									except:  # noqa
													# Something did go wrong, so decrement the counter.
													with self._thread_lock:
																	SCREAMING_SNAKE_CASE__		  =				max(0   , self._lock_counter - 1		)

													raise
									return _Acquire_ReturnProxy(lock=self		)



					def 			lowerCAmelCase_					(						self			:				str   , _snake_case			:				Tuple=False		)							->       Dict:






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

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

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

									return None



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



					def __exit__(						self			:				int   , _snake_case			:				Dict   , _snake_case			:				Dict   , _snake_case			:				Optional[int]		)							->       Optional[Any]:
									self.release()
									return None



					def __del__(						self			:				Any		)							->       Tuple:
									self.release(force=__UpperCamelCase		)
									return None



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



class lowerCamelCase		(_UpperCAmelCase      ):






					'''simple docstring'''


					def __init__(						self			:				Optional[Any]   , _snake_case			:				Optional[Any]   , _snake_case			:				Tuple=-1   , _snake_case			:				Dict=None		)							->       int:
									from .file_utils import relative_to_absolute_path

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



					def 			lowerCAmelCase_					(						self			:				List[str]		)							->       List[Any]:
									SCREAMING_SNAKE_CASE__		  =				os.O_RDWR | os.O_CREAT | os.O_TRUNC

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



					def 			lowerCAmelCase_					(						self			:				Tuple		)							->       Tuple:
									SCREAMING_SNAKE_CASE__		  =				self._lock_file_fd
									SCREAMING_SNAKE_CASE__		  =				None
									msvcrt.locking(__UpperCamelCase   , msvcrt.LK_UNLCK   , 1		)
									os.close(__UpperCamelCase		)

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



class lowerCamelCase		(_UpperCAmelCase      ):






					'''simple docstring'''


					def __init__(						self			:				Any   , _snake_case			:				Any   , _snake_case			:				Tuple=-1   , _snake_case			:				Union[str, Any]=None		)							->       Any:
									SCREAMING_SNAKE_CASE__		  =				os.statvfs(os.path.dirname(__UpperCamelCase		)		).f_namemax
									super().__init__(__UpperCamelCase   , timeout=__UpperCamelCase   , max_filename_length=__UpperCamelCase		)



					def 			lowerCAmelCase_					(						self			:				Optional[Any]		)							->       Tuple:
									SCREAMING_SNAKE_CASE__		  =				os.O_RDWR | os.O_CREAT | os.O_TRUNC
									SCREAMING_SNAKE_CASE__		  =				os.open(self._lock_file   , __UpperCamelCase		)

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



					def 			lowerCAmelCase_					(						self			:				List[str]		)							->       Optional[Any]:
									SCREAMING_SNAKE_CASE__		  =				self._lock_file_fd
									SCREAMING_SNAKE_CASE__		  =				None
									fcntl.flock(__UpperCamelCase   , fcntl.LOCK_UN		)
									os.close(__UpperCamelCase		)
									return None



class lowerCamelCase		(_UpperCAmelCase      ):






					'''simple docstring'''


					def 			lowerCAmelCase_					(						self			:				List[Any]		)							->       Optional[Any]:
									SCREAMING_SNAKE_CASE__		  =				os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
									try:
													SCREAMING_SNAKE_CASE__		  =				os.open(self._lock_file   , __UpperCamelCase		)
									except OSError:
													pass
									else:
													SCREAMING_SNAKE_CASE__		  =				fd
									return None



					def 			lowerCAmelCase_					(						self			:				str		)							->       List[str]:
									os.close(self._lock_file_fd		)
									SCREAMING_SNAKE_CASE__		  =				None

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



_A   					=	None

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

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





"""simple docstring"""






def 						SCREAMING_SNAKE_CASE					(  __UpperCAmelCase       )							->   int:

				if not isinstance(__UpperCAmelCase  ,					__UpperCAmelCase       ):
								raise ValueError("multiplicative_persistence() only accepts integral values"       )
				if num < 0:
								raise ValueError("multiplicative_persistence() does not accept negative values"       )

				SCREAMING_SNAKE_CASE__		  =				0
				SCREAMING_SNAKE_CASE__		  =				str(__UpperCAmelCase       )

				while len(__UpperCAmelCase       ) != 1:
								SCREAMING_SNAKE_CASE__		  =				[int(__UpperCAmelCase       ) for i in num_string]

								SCREAMING_SNAKE_CASE__		  =				1
								for i in range(0  ,					len(__UpperCAmelCase       )       ):
												total *= numbers[i]

								SCREAMING_SNAKE_CASE__		  =				str(__UpperCAmelCase       )

								steps += 1
				return steps




def 						SCREAMING_SNAKE_CASE					(  __UpperCAmelCase       )							->   int:

				if not isinstance(__UpperCAmelCase  ,					__UpperCAmelCase       ):
								raise ValueError("additive_persistence() only accepts integral values"       )
				if num < 0:
								raise ValueError("additive_persistence() does not accept negative values"       )

				SCREAMING_SNAKE_CASE__		  =				0
				SCREAMING_SNAKE_CASE__		  =				str(__UpperCAmelCase       )

				while len(__UpperCAmelCase       ) != 1:
								SCREAMING_SNAKE_CASE__		  =				[int(__UpperCAmelCase       ) for i in num_string]

								SCREAMING_SNAKE_CASE__		  =				0
								for i in range(0  ,					len(__UpperCAmelCase       )       ):
												total += numbers[i]

								SCREAMING_SNAKE_CASE__		  =				str(__UpperCAmelCase       )

								steps += 1
				return steps


if __name__ == "__main__":
						import doctest

						doctest.testmod()




