






'''simple docstring'''



import logging
import os
import threading
import time


try:
      import warnings
except ImportError:
      SCREAMING_SNAKE_CASE							=						None

try:
      import msvcrt
except ImportError:
      SCREAMING_SNAKE_CASE							=						None

try:
      import fcntl
except ImportError:
      SCREAMING_SNAKE_CASE							=						None


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


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

SCREAMING_SNAKE_CASE							=						'3.0.12'


SCREAMING_SNAKE_CASE							=						None


def 				lowercase_					(     )							->   int:



 """simple docstring"""




 global _logger
 lowercase       : List[str]						=_logger or logging.getLogger(__name__      )
 return _logger
class UpperCAmelCase_       (     _a					):




      """simple docstring"""




      def __init__(			self      :		Optional[Any]  ,   UpperCAmelCase      :		List[str]		)	->	Optional[Any]:


       '''simple docstring'''

       lowercase       : List[str]						=lock_file
       return None






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


       '''simple docstring'''

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





class UpperCAmelCase_       :




      """simple docstring"""




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


       '''simple docstring'''

       lowercase       : Tuple						=lock
       return None






      def __enter__(			self      :		int		)	->	int:


       '''simple docstring'''

       return self.lock






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


       '''simple docstring'''

       self.lock.release()
       return None





class UpperCAmelCase_       :




      """simple docstring"""




      def __init__(			self      :		List[Any]  ,   UpperCAmelCase      :		Dict  ,   UpperCAmelCase      :		List[Any]=-1  ,   UpperCAmelCase      :		List[Any]=None		)	->	List[Any]:


       '''simple docstring'''

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

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

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






      @property
      def 					A__   (			self      :		Dict		)	->	Optional[int]:


       '''simple docstring'''

       return self._lock_file






      @property
      def 					A__   (			self      :		List[str]		)	->	Any:


       '''simple docstring'''

       return self._timeout






      @timeout.setter
      def 					A__   (			self      :		Any  ,   UpperCAmelCase      :		List[str]		)	->	Any:


       '''simple docstring'''

       lowercase       : Dict						=float(snake_case_		)
       return None






      def 					A__   (			self      :		str		)	->	Dict:


       '''simple docstring'''

       raise NotImplementedError()






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


       '''simple docstring'''

       raise NotImplementedError()






      @property
      def 					A__   (			self      :		Any		)	->	List[str]:


       '''simple docstring'''

       return self._lock_file_fd is not None






      def 					A__   (			self      :		Tuple  ,   UpperCAmelCase      :		List[str]=None  ,   UpperCAmelCase      :		Any=0.0_5		)	->	Optional[Any]:


       '''simple docstring'''

       if timeout is None:
        lowercase       : Union[str, 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

       lowercase       : Dict						=id(self		)
       lowercase       : Dict						=self._lock_file
       lowercase       : Dict						=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(snake_case_		)
       except:  # noqa
        # Something did go wrong, so decrement the counter.
        with self._thread_lock:
         lowercase       : Tuple						=max(0  ,   self._lock_counter - 1		)

        raise
       return _Acquire_ReturnProxy(lock=self		)






      def 					A__   (			self      :		Union[str, Any]  ,   UpperCAmelCase      :		List[str]=False		)	->	Tuple:


       '''simple docstring'''

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

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

          logger().debug(f'Attempting to release lock {lock_id} on {lock_filename}'		)
          self._release()
          lowercase       : Union[str, Any]						=0
          logger().debug(f'Lock {lock_id} released on {lock_filename}'		)

       return None






      def __enter__(			self      :		int		)	->	str:


       '''simple docstring'''

       self.acquire()
       return self






      def __exit__(			self      :		int  ,   UpperCAmelCase      :		Any  ,   UpperCAmelCase      :		Tuple  ,   UpperCAmelCase      :		Optional[Any]		)	->	Any:


       '''simple docstring'''

       self.release()
       return None






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


       '''simple docstring'''

       self.release(force=snake_case_		)
       return None






      def 					A__   (			self      :		str  ,   UpperCAmelCase      :		str  ,   UpperCAmelCase      :		int		)	->	Tuple:


       '''simple docstring'''

       lowercase       : str						=os.path.basename(snake_case_		)
       if len(snake_case_		) > max_length and max_length > 0:
        lowercase       : List[str]						=os.path.dirname(snake_case_		)
        lowercase       : str						=str(hash(snake_case_		)		)
        lowercase       : List[str]						=filename[: max_length - len(snake_case_		) - 8] + """...""" + hashed_filename + """.lock"""
        return os.path.join(snake_case_  ,   snake_case_		)
       else:
        return path





class UpperCAmelCase_       (     _a					):




      """simple docstring"""




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


       '''simple docstring'''

       from .file_utils import relative_to_absolute_path

       super().__init__(snake_case_  ,   timeout=snake_case_  ,   max_filename_length=snake_case_		)
       lowercase       : Optional[int]						="""\\\\?\\""" + relative_to_absolute_path(self.lock_file		)






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


       '''simple docstring'''

       lowercase       : List[str]						=os.O_RDWR | os.O_CREAT | os.O_TRUNC

       try:
        lowercase       : Optional[Any]						=os.open(self._lock_file  ,   snake_case_		)
       except OSError:
        pass
       else:
        try:
         msvcrt.locking(snake_case_  ,   msvcrt.LK_NBLCK  ,   1		)
        except OSError:
         os.close(snake_case_		)
        else:
         lowercase       : Tuple						=fd
       return None






      def 					A__   (			self      :		Tuple		)	->	Optional[Any]:


       '''simple docstring'''

       lowercase       : Optional[int]						=self._lock_file_fd
       lowercase       : Dict						=None
       msvcrt.locking(snake_case_  ,   msvcrt.LK_UNLCK  ,   1		)
       os.close(snake_case_		)

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





class UpperCAmelCase_       (     _a					):




      """simple docstring"""




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


       '''simple docstring'''

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






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


       '''simple docstring'''

       lowercase       : Optional[Any]						=os.O_RDWR | os.O_CREAT | os.O_TRUNC
       lowercase       : Any						=os.open(self._lock_file  ,   snake_case_		)

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






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


       '''simple docstring'''

       lowercase       : Optional[Any]						=self._lock_file_fd
       lowercase       : Tuple						=None
       fcntl.flock(snake_case_  ,   fcntl.LOCK_UN		)
       os.close(snake_case_		)
       return None





class UpperCAmelCase_       (     _a					):




      """simple docstring"""




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


       '''simple docstring'''

       lowercase       : int						=os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_TRUNC
       try:
        lowercase       : Tuple						=os.open(self._lock_file  ,   snake_case_		)
       except OSError:
        pass
       else:
        lowercase       : Union[str, Any]						=fd
       return None






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


       '''simple docstring'''

       os.close(self._lock_file_fd		)
       lowercase       : Any						=None

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





SCREAMING_SNAKE_CASE							=						None

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

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








'''simple docstring'''





import importlib
import shutil
import threading
import warnings
from typing import List

import fsspec
import fsspec.asyn

from . import compression
from .hffilesystem import HfFileSystem


__a  			=						importlib.util.find_spec("s3fs") is not None

if _has_safs:
       from .safilesystem import SaFileSystem  # noqa: F401

__a  			=						[
    compression.BzaFileSystem,
    compression.GzipFileSystem,
    compression.LzaFileSystem,
    compression.XzFileSystem,
    compression.ZstdFileSystem,
]

# Register custom filesystems
for fs_class in COMPRESSION_FILESYSTEMS + [HfFileSystem]:
       if fs_class.protocol in fsspec.registry and fsspec.registry[fs_class.protocol] is not fs_class:
              warnings.warn(F"A filesystem protocol was already set for {fs_class.protocol} and will be overwritten.")
       fsspec.register_implementation(fs_class.protocol, fs_class, clobber=True)





def 		__snake_case(							_lowerCAmelCase						)						->					str:
      if "://" in dataset_path:
            snake_case__    :       Tuple       		= dataset_path.split("""://"""						)[1]
      return dataset_path





def 		__snake_case(							_lowerCAmelCase						)						->					bool:
      if fs is not None and fs.protocol != "file":
            return True
      else:
            return False





def 		__snake_case(							_lowerCAmelCase	,  _lowerCAmelCase	,  _lowerCAmelCase						)						->					Optional[int]:
      snake_case__    :       int       		= not is_remote_filesystem(_lowerCAmelCase						)
      if is_local:
            # LocalFileSystem.mv does copy + rm, it is more efficient to simply move a local directory
            shutil.move(fs._strip_protocol(_lowerCAmelCase						)	,  fs._strip_protocol(_lowerCAmelCase						)						)
      else:
            fs.mv(_lowerCAmelCase	,  _lowerCAmelCase	,  recursive=_lowerCAmelCase						)





def 		__snake_case(							)						->					None:
      if hasattr(fsspec.asyn	,  """reset_lock"""						):
            # for future fsspec>2022.05.0
            fsspec.asyn.reset_lock()
      else:
            snake_case__    :       Dict       		= None
            snake_case__    :       List[str]       		= None
            snake_case__    :       Union[str, Any]       		= threading.Lock()
