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

import fsspec
import fsspec.asyn

from . import compression
from .hffilesystem import HfFileSystem


lowercase  : Any						   =							importlib.util.find_spec("""s3fs""") is not None

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

lowercase  : List[compression.BaseCompressedFileFileSystem]						   =							[
    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(  SCREAMING_SNAKE_CASE__    ):
       if "://" in dataset_path:
              lowercase							:					Tuple	     =				dataset_path.split("""://"""    )[1]
       return dataset_path







def 						_snake_case(  SCREAMING_SNAKE_CASE__    ):
       if fs is not None and fs.protocol != "file":
              return True
       else:
              return False







def 						_snake_case(  SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__    ):
       lowercase							:					Optional[Any]	     =				not is_remote_filesystem(SCREAMING_SNAKE_CASE__    )
       if is_local:
              # LocalFileSystem.mv does copy + rm, it is more efficient to simply move a local directory
              shutil.move(fs._strip_protocol(SCREAMING_SNAKE_CASE__    ) ,     fs._strip_protocol(SCREAMING_SNAKE_CASE__    )    )
       else:
              fs.mv(SCREAMING_SNAKE_CASE__ ,     SCREAMING_SNAKE_CASE__ ,     recursive=SCREAMING_SNAKE_CASE__    )







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


from ...processing_utils import ProcessorMixin

class __snake_case		(				lowerCAmelCase					):
    _a		:						Union[str, Any]=    "WhisperFeatureExtractor"
    _a		:						int=    "WhisperTokenizer"
    def __init__(  self		,snake_case		,snake_case				):



           '''simple docstring'''







           super().__init__(snake_case		,snake_case				)
           lowercase							:					Optional[int]	     =				self.feature_extractor
           lowercase							:					Tuple	     =				False
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case=None		,snake_case=None		,snake_case=True				):



           '''simple docstring'''







           return self.tokenizer.get_decoder_prompt_ids(task=snake_case		,language=snake_case		,no_timestamps=snake_case				)
    def __call__(  self		,*snake_case		,**snake_case				):



           '''simple docstring'''







           if self._in_target_context_manager:
                  return self.current_processor(*snake_case		,**snake_case				)

           lowercase							:					Optional[Any]	     =				kwargs.pop("""audio"""		,snake_case				)
           lowercase							:					str	     =				kwargs.pop("""sampling_rate"""		,snake_case				)
           lowercase							:					Dict	     =				kwargs.pop("""text"""		,snake_case				)
           if len(snake_case				) > 0:
                  lowercase							:					List[Any]	     =				args[0]
                  lowercase							:					Tuple	     =				args[1:]

           if audio is None and text is None:
                  raise ValueError("""You need to specify either an `audio` or `text` input to process."""				)

           if audio is not None:
                  lowercase							:					Any	     =				self.feature_extractor(snake_case		,*snake_case		,sampling_rate=snake_case		,**snake_case				)
           if text is not None:
                  lowercase							:					str	     =				self.tokenizer(snake_case		,**snake_case				)

           if text is None:
                  return inputs

           elif audio is None:
                  return encodings
           else:
                  lowercase							:					List[Any]	     =				encodings["""input_ids"""]
                  return inputs
    def 				_SCREAMING_SNAKE_CASE	(  self		,*snake_case		,**snake_case				):



           '''simple docstring'''







           return self.tokenizer.batch_decode(*snake_case		,**snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,*snake_case		,**snake_case				):



           '''simple docstring'''







           return self.tokenizer.decode(*snake_case		,**snake_case				)
    def 				_SCREAMING_SNAKE_CASE	(  self		,snake_case		,snake_case="np"				):



           '''simple docstring'''







           return self.tokenizer.get_prompt_ids(snake_case		,return_tensors=snake_case				)
