




import logging
import os
import threading
import time


try:
 import warnings
except ImportError:
 lowerCAmelCase_		      = None

try:
 import msvcrt
except ImportError:
 lowerCAmelCase_		      = None

try:
 import fcntl
except ImportError:
 lowerCAmelCase_		      = None


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


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

lowerCAmelCase_		      = '3.0.12'


lowerCAmelCase_		      = None






def 					snake_case(				) ->							Dict:





     '''simple docstring'''


     global _logger
     lowercase					:   str								= _logger or logging.getLogger(__name__    )
     return _logger
class 						_A					(  _UpperCAmelCase      ):






  def __init__(  self		:	Union[str, Any]      ,						_A		:	int      )		->    Union[str, Any]:






       """simple docstring"""







       lowercase					:   Any								= lock_file
       return None






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






       """simple docstring"""







       lowercase					:   Optional[Any]								= f"""The file lock '{self.lock_file}' could not be acquired."""
       return temp


class 						_A					:






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






       """simple docstring"""







       lowercase					:   List[Any]								= lock
       return None






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






       """simple docstring"""







       return self.lock






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






       """simple docstring"""







       self.lock.release()
       return None


class 						_A					:






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






       """simple docstring"""







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

       # The default timeout value.
       lowercase					:   List[Any]								= timeout

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






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






       """simple docstring"""







       return self._lock_file






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






       """simple docstring"""







       return self._timeout






  @timeout.setter
  def 					__a			(  self		:	Optional[Any]      ,						_A		:	Optional[Any]      )		->    Dict:






       """simple docstring"""







       lowercase					:   Dict								= float(a__      )
       return None






  def 					__a			(  self		:	Optional[int]      )		->    Dict:






       """simple docstring"""







       raise NotImplementedError()






  def 					__a			(  self		:	Optional[int]      )		->    List[str]:






       """simple docstring"""







       raise NotImplementedError()






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






       """simple docstring"""







       return self._lock_file_fd is not None






  def 					__a			(  self		:	Dict      ,						_A		:	Any=None      ,						_A		:	Optional[Any]=0.05      )		->    Tuple:






       """simple docstring"""







       if timeout is None:
            lowercase					:   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

       lowercase					:   Union[str, Any]								= id(self      )
       lowercase					:   Any								= self._lock_file
       lowercase					:   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(a__      )
       except:  # noqa
            # Something did go wrong, so decrement the counter.
            with self._thread_lock:
                 lowercase					:   List[Any]								= max(0      ,						self._lock_counter - 1      )

            raise
       return _Acquire_ReturnProxy(lock=self      )






  def 					__a			(  self		:	List[Any]      ,						_A		:	Union[str, Any]=False      )		->    Union[str, Any]:






       """simple docstring"""







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

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

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

       return None






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






       """simple docstring"""







       self.acquire()
       return self






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






       """simple docstring"""







       self.release()
       return None






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






       """simple docstring"""







       self.release(force=a__      )
       return None






  def 					__a			(  self		:	Dict      ,						_A		:	str      ,						_A		:	int      )		->    Any:






       """simple docstring"""







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


class 						_A					(  _UpperCAmelCase      ):






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






       """simple docstring"""







       from .file_utils import relative_to_absolute_path

       super().__init__(a__      ,						timeout=a__      ,						max_filename_length=a__      )
       lowercase					:   str								= '''\\\\?\\''' + relative_to_absolute_path(self.lock_file      )






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






       """simple docstring"""







       lowercase					:   int								= os.O_RDWR | os.O_CREAT | os.O_TRUNC

       try:
            lowercase					:   int								= os.open(self._lock_file      ,						a__      )
       except OSError:
            pass
       else:
            try:
                 msvcrt.locking(a__      ,						msvcrt.LK_NBLCK      ,						1      )
            except OSError:
                 os.close(a__      )
            else:
                 lowercase					:   Any								= fd
       return None






  def 					__a			(  self		:	int      )		->    Any:






       """simple docstring"""







       lowercase					:   int								= self._lock_file_fd
       lowercase					:   Dict								= 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 						_A					(  _UpperCAmelCase      ):






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






       """simple docstring"""







       lowercase					:   Dict								= os.statvfs(os.path.dirname(a__      )      ).f_namemax
       super().__init__(a__      ,						timeout=a__      ,						max_filename_length=a__      )






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






       """simple docstring"""







       lowercase					:   Optional[int]								= os.O_RDWR | os.O_CREAT | os.O_TRUNC
       lowercase					:   str								= os.open(self._lock_file      ,						a__      )

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






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






       """simple docstring"""







       lowercase					:   Any								= self._lock_file_fd
       lowercase					:   List[str]								= None
       fcntl.flock(a__      ,						fcntl.LOCK_UN      )
       os.close(a__      )
       return None


class 						_A					(  _UpperCAmelCase      ):






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






       """simple docstring"""







       lowercase					:   Optional[Any]								= os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
       try:
            lowercase					:   Any								= os.open(self._lock_file      ,						a__      )
       except OSError:
            pass
       else:
            lowercase					:   Dict								= fd
       return None






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






       """simple docstring"""







       os.close(self._lock_file_fd      )
       lowercase					:   Optional[int]								= None

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


lowerCAmelCase_		      = None

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

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






import os
import unittest

from huggingface_hub.utils import are_progress_bars_disabled

import transformers.models.bart.tokenization_bart
from transformers import logging
from transformers.testing_utils import CaptureLogger, mockenv, mockenv_context
from transformers.utils.logging import disable_progress_bar, enable_progress_bar

class 		SCREAMING_SNAKE_CASE__				(       unittest.TestCase  ):


      def    a (self       :      Dict      ):





         """simple docstring"""



         __snake_case							       =   logging.get_logger()

         # the current default level is logging.WARNING
         __snake_case							       =   logging.get_verbosity()

         logging.set_verbosity_error()
         self.assertEqual(logger.getEffectiveLevel()				,			logging.get_verbosity()      )

         logging.set_verbosity_warning()
         self.assertEqual(logger.getEffectiveLevel()				,			logging.get_verbosity()      )

         logging.set_verbosity_info()
         self.assertEqual(logger.getEffectiveLevel()				,			logging.get_verbosity()      )

         logging.set_verbosity_debug()
         self.assertEqual(logger.getEffectiveLevel()				,			logging.get_verbosity()      )

         # restore to the original level
         logging.set_verbosity(a__      )


      def    a (self       :      Dict      ):





         """simple docstring"""



         __snake_case							       =   logging.get_verbosity()

         __snake_case							       =   logging.get_logger('''transformers.models.bart.tokenization_bart'''      )
         __snake_case							       =   '''Testing 1, 2, 3'''

         # should be able to log warnings (if default settings weren't overridden by `pytest --log-level-all`)
         if level_origin <= logging.WARNING:
            with CaptureLogger(a__      ) as cl:
               logger.warning(a__      )
            self.assertEqual(cl.out				,			msg + '''\n'''      )

         # this is setting the level for all of `transformers.*` loggers
         logging.set_verbosity_error()

         # should not be able to log warnings
         with CaptureLogger(a__      ) as cl:
            logger.warning(a__      )
         self.assertEqual(cl.out				,			''''''      )

         # should be able to log warnings again
         logging.set_verbosity_warning()
         with CaptureLogger(a__      ) as cl:
            logger.warning(a__      )
         self.assertEqual(cl.out				,			msg + '''\n'''      )

         # restore to the original level
         logging.set_verbosity(a__      )


      @mockenv(TRANSFORMERS_VERBOSITY='''error'''      )
      def    a (self       :      Dict      ):





         """simple docstring"""



         transformers.utils.logging._reset_library_root_logger()
         # this action activates the env var
         __snake_case							       =   logging.get_logger('''transformers.models.bart.tokenization_bart'''      )

         __snake_case							       =   os.getenv('''TRANSFORMERS_VERBOSITY'''				,			a__      )
         __snake_case							       =   logging.log_levels[env_level_str]

         __snake_case							       =   logging.get_verbosity()
         self.assertEqual(
             a__				,			a__				,			f"""TRANSFORMERS_VERBOSITY={env_level_str}/{env_level}, but internal verbosity is {current_level}"""				,			)

         # restore to the original level
         __snake_case							       =   ''''''
         transformers.utils.logging._reset_library_root_logger()


      @mockenv(TRANSFORMERS_VERBOSITY='''super-error'''      )
      def    a (self       :      List[Any]      ):





         """simple docstring"""



         transformers.utils.logging._reset_library_root_logger()
         __snake_case							       =   logging.logging.getLogger()
         with CaptureLogger(a__      ) as cl:
            # this action activates the env var
            logging.get_logger('''transformers.models.bart.tokenization_bart'''      )
         self.assertIn('''Unknown option TRANSFORMERS_VERBOSITY=super-error'''				,			cl.out      )

         # no need to restore as nothing was changed


      def    a (self       :      Any      ):





         """simple docstring"""



         transformers.utils.logging._reset_library_root_logger()

         __snake_case							       =   logging.get_logger('''transformers.models.bart.tokenization_bart'''      )
         __snake_case							       =   '''Testing 1, 2, 3'''

         with mockenv_context(TRANSFORMERS_NO_ADVISORY_WARNINGS='''1'''      ):
            # nothing should be logged as env var disables this method
            with CaptureLogger(a__      ) as cl:
               logger.warning_advice(a__      )
            self.assertEqual(cl.out				,			''''''      )

         with mockenv_context(TRANSFORMERS_NO_ADVISORY_WARNINGS=''''''      ):
            # should log normally as TRANSFORMERS_NO_ADVISORY_WARNINGS is unset
            with CaptureLogger(a__      ) as cl:
               logger.warning_advice(a__      )
            self.assertEqual(cl.out				,			msg + '''\n'''      )






def lowerCamelCase__    (							)      ->				str:
   disable_progress_bar()
   assert are_progress_bars_disabled()

   enable_progress_bar()
   assert not are_progress_bars_disabled()

