

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      _lowerCAmelCase(	):
  global _logger
  UpperCAmelCase					      =  _logger or logging.getLogger(__name__			)
  return _logger






class 						__magic_name__    (       __UpperCAmelCase    ):


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


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






class 						__magic_name__    :


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


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


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






class 						__magic_name__    :


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

   # The default timeout value.
   UpperCAmelCase					      =  timeout

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


 @property
 def _UpperCamelCase	(						self		: Optional[Any]    )    ->      List[str]:
   return self._lock_file


 @property
 def _UpperCamelCase	(						self		: str    )    ->      Optional[int]:
   return self._timeout


 @timeout.setter
 def _UpperCamelCase	(						self		: Optional[int]			,						lowerCAmelCase__		: List[str]    )    ->      List[str]:
   UpperCAmelCase					      =  float(lowerCAmelCase__    )
   return None


 def _UpperCamelCase	(						self		: Union[str, Any]    )    ->      List[Any]:
   raise NotImplementedError()


 def _UpperCamelCase	(						self		: Any    )    ->      Optional[Any]:
   raise NotImplementedError()


 @property
 def _UpperCamelCase	(						self		: Any    )    ->      List[Any]:
   return self._lock_file_fd is not None


 def _UpperCamelCase	(						self		: Optional[int]			,						lowerCAmelCase__		: Any=None			,						lowerCAmelCase__		: str=0.05    )    ->      Union[str, Any]:

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

   UpperCAmelCase					      =  id(self    )
   UpperCAmelCase					      =  self._lock_file
   UpperCAmelCase					      =  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:
       UpperCAmelCase					      =  max(0			,						self._lock_counter - 1    )

     raise
   return _Acquire_ReturnProxy(lock=self    )


 def _UpperCamelCase	(						self		: Union[str, Any]			,						lowerCAmelCase__		: int=False    )    ->      Optional[int]:

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

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

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

   return None


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


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


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


 def _UpperCamelCase	(						self		: List[Any]			,						lowerCAmelCase__		: str			,						lowerCAmelCase__		: str    )    ->      Any:
   UpperCAmelCase					      =  os.path.basename(lowerCAmelCase__    )
   if len(lowerCAmelCase__    ) > max_length and max_length > 0:
     UpperCAmelCase					      =  os.path.dirname(lowerCAmelCase__    )
     UpperCAmelCase					      =  str(hash(lowerCAmelCase__    )    )
     UpperCAmelCase					      =  filename[: max_length - len(lowerCAmelCase__    ) - 8] + "..." + hashed_filename + ".lock"
     return os.path.join(lowerCAmelCase__			,						lowerCAmelCase__    )
   else:
     return path






class 						__magic_name__    (       __UpperCAmelCase    ):


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

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


 def _UpperCamelCase	(						self		: int    )    ->      Optional[int]:
   UpperCAmelCase					      =  os.O_RDWR | os.O_CREAT | os.O_TRUNC

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


 def _UpperCamelCase	(						self		: List[Any]    )    ->      Dict:
   UpperCAmelCase					      =  self._lock_file_fd
   UpperCAmelCase					      =  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 						__magic_name__    (       __UpperCAmelCase    ):


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


 def _UpperCamelCase	(						self		: Any    )    ->      str:
   UpperCAmelCase					      =  os.O_RDWR | os.O_CREAT | os.O_TRUNC
   UpperCAmelCase					      =  os.open(self._lock_file			,						lowerCAmelCase__    )

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


 def _UpperCamelCase	(						self		: Tuple    )    ->      List[Any]:
   UpperCAmelCase					      =  self._lock_file_fd
   UpperCAmelCase					      =  None
   fcntl.flock(lowerCAmelCase__			,						fcntl.LOCK_UN    )
   os.close(lowerCAmelCase__    )
   return None






class 						__magic_name__    (       __UpperCAmelCase    ):


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


 def _UpperCamelCase	(						self		: List[Any]    )    ->      int:
   os.close(self._lock_file_fd    )
   UpperCAmelCase					      =  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 sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoFeatureExtractor, WavaVecaFeatureExtractor
from transformers.testing_utils import TOKEN, USER, get_tests_dir, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / "utils"))

from test_module.custom_feature_extraction import CustomFeatureExtractor  # noqa E402


lowerCAmelCase__       = get_tests_dir("fixtures")






class 						__magic_name__    (       unittest.TestCase    ):


 def _UpperCamelCase	(						self		: Tuple    )    ->      Union[str, Any]:
   # A mock response for an HTTP head request to emulate server down
   UpperCAmelCase					      =  mock.Mock()
   UpperCAmelCase					      =  5_0_0
   UpperCAmelCase					      =  {}
   UpperCAmelCase					      =  HTTPError
   UpperCAmelCase					      =  {}

   # Download this model to make sure it's in the cache.
   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"    )
   # Under the mock environment we get a 500 error when trying to reach the model.
   with mock.patch("requests.Session.request"			,						return_value=lowerCAmelCase__    ) as mock_head:
     UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained("hf-internal-testing/tiny-random-wav2vec2"    )
     # This check we did call the fake head request
     mock_head.assert_called()


 def _UpperCamelCase	(						self		: List[Any]    )    ->      Dict:
   # This test is for deprecated behavior and can be removed in v5
   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(
       "https://huggingface.co/hf-internal-testing/tiny-random-wav2vec2/resolve/main/preprocessor_config.json"    )






@is_staging_test
class 						__magic_name__    (       unittest.TestCase    ):


 @classmethod
 def _UpperCamelCase	(						cls		: List[str]    )    ->      List[Any]:
   UpperCAmelCase					      =  TOKEN
   HfFolder.save_token(lowerCAmelCase__    )


 @classmethod
 def _UpperCamelCase	(						cls		: Optional[int]    )    ->      Union[str, Any]:
   try:
     delete_repo(token=cls._token			,						repo_id="test-feature-extractor"    )
   except HTTPError:
     pass

   try:
     delete_repo(token=cls._token			,						repo_id="valid_org/test-feature-extractor-org"    )
   except HTTPError:
     pass

   try:
     delete_repo(token=cls._token			,						repo_id="test-dynamic-feature-extractor"    )
   except HTTPError:
     pass


 def _UpperCamelCase	(						self		: Any    )    ->      Any:
   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase__    )
   feature_extractor.push_to_hub("test-feature-extractor"			,						use_auth_token=self._token    )

   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(f"{USER}/test-feature-extractor"    )
   for k, v in feature_extractor.__dict__.items():
     self.assertEqual(lowerCAmelCase__			,						getattr(lowerCAmelCase__			,						lowerCAmelCase__    )    )

   # Reset repo
   delete_repo(token=self._token			,						repo_id="test-feature-extractor"    )

   # Push to hub via save_pretrained
   with tempfile.TemporaryDirectory() as tmp_dir:
     feature_extractor.save_pretrained(
         lowerCAmelCase__			,						repo_id="test-feature-extractor"			,						push_to_hub=lowerCAmelCase__			,						use_auth_token=self._token    )

   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(f"{USER}/test-feature-extractor"    )
   for k, v in feature_extractor.__dict__.items():
     self.assertEqual(lowerCAmelCase__			,						getattr(lowerCAmelCase__			,						lowerCAmelCase__    )    )


 def _UpperCamelCase	(						self		: List[Any]    )    ->      Tuple:
   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained(lowerCAmelCase__    )
   feature_extractor.push_to_hub("valid_org/test-feature-extractor"			,						use_auth_token=self._token    )

   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor"    )
   for k, v in feature_extractor.__dict__.items():
     self.assertEqual(lowerCAmelCase__			,						getattr(lowerCAmelCase__			,						lowerCAmelCase__    )    )

   # Reset repo
   delete_repo(token=self._token			,						repo_id="valid_org/test-feature-extractor"    )

   # Push to hub via save_pretrained
   with tempfile.TemporaryDirectory() as tmp_dir:
     feature_extractor.save_pretrained(
         lowerCAmelCase__			,						repo_id="valid_org/test-feature-extractor-org"			,						push_to_hub=lowerCAmelCase__			,						use_auth_token=self._token    )

   UpperCAmelCase					      =  WavaVecaFeatureExtractor.from_pretrained("valid_org/test-feature-extractor-org"    )
   for k, v in feature_extractor.__dict__.items():
     self.assertEqual(lowerCAmelCase__			,						getattr(lowerCAmelCase__			,						lowerCAmelCase__    )    )


 def _UpperCamelCase	(						self		: Dict    )    ->      List[str]:
   CustomFeatureExtractor.register_for_auto_class()
   UpperCAmelCase					      =  CustomFeatureExtractor.from_pretrained(lowerCAmelCase__    )

   feature_extractor.push_to_hub("test-dynamic-feature-extractor"			,						use_auth_token=self._token    )

   # This has added the proper auto_map field to the config
   self.assertDictEqual(
       feature_extractor.auto_map			,						{"AutoFeatureExtractor": "custom_feature_extraction.CustomFeatureExtractor"}			,						)

   UpperCAmelCase					      =  AutoFeatureExtractor.from_pretrained(
       f"{USER}/test-dynamic-feature-extractor"			,						trust_remote_code=lowerCAmelCase__    )
   # Can't make an isinstance check because the new_feature_extractor is from the CustomFeatureExtractor class of a dynamic module
   self.assertEqual(new_feature_extractor.__class__.__name__			,						"CustomFeatureExtractor"    )



