

'''simple docstring'''

import importlib
import os
import platform
from pathlib import Path

from packaging import version

from .utils.logging import get_logger


A_      =							get_logger(__name__)

# Datasets
A_      =							"https://s3.amazonaws.com/datasets.huggingface.co/datasets/datasets"
A_      =							"https://cdn-datasets.huggingface.co/datasets/datasets"
A_      =							"https://raw.githubusercontent.com/huggingface/datasets/{revision}/datasets/{path}/{name}"

# Metrics
A_      =							"https://s3.amazonaws.com/datasets.huggingface.co/datasets/metrics"
A_      =							"https://cdn-datasets.huggingface.co/datasets/metric"
A_      =							"https://raw.githubusercontent.com/huggingface/datasets/{revision}/metrics/{path}/{name}"

# Hub
A_      =							os.environ.get("HF_ENDPOINT", "https://huggingface.co")
A_      =							HF_ENDPOINT + "/datasets/{repo_id}/resolve/{revision}/{path}"
A_      =							"main"

A_      =							version.parse(platform.python_version())

if PY_VERSION < version.parse("3.8"):
    import importlib_metadata
else:
    import importlib.metadata as importlib_metadata

# General environment variables accepted values for booleans
A_      =							{"1", "ON", "YES", "TRUE"}
A_      =							ENV_VARS_TRUE_VALUES.union({"AUTO"})


# Imports
A_      =							version.parse(importlib_metadata.version("dill"))
A_      =							version.parse(importlib_metadata.version("pandas"))
A_      =							version.parse(importlib_metadata.version("pyarrow"))

A_      =							os.environ.get("USE_TF", "AUTO").upper()
A_      =							os.environ.get("USE_TORCH", "AUTO").upper()
A_      =							os.environ.get("USE_JAX", "AUTO").upper()

A_      =							"N/A"
A_      =							False

if USE_TORCH in ENV_VARS_TRUE_AND_AUTO_VALUES and USE_TF not in ENV_VARS_TRUE_VALUES:
    A_      =							importlib.util.find_spec("torch") is not None
    if TORCH_AVAILABLE:
        try:
            A_      =							version.parse(importlib_metadata.version("torch"))
            logger.info(f'''PyTorch version {TORCH_VERSION} available.''')
        except importlib_metadata.PackageNotFoundError:
            pass
else:
    logger.info("Disabling PyTorch because USE_TF is set")

A_      =							"N/A"
A_      =							False

if USE_TF in ENV_VARS_TRUE_AND_AUTO_VALUES and USE_TORCH not in ENV_VARS_TRUE_VALUES:
    A_      =							importlib.util.find_spec("tensorflow") is not None
    if TF_AVAILABLE:
        # For the metadata, we have to look for both tensorflow and tensorflow-cpu
        for package in [
            "tensorflow",
            "tensorflow-cpu",
            "tensorflow-gpu",
            "tf-nightly",
            "tf-nightly-cpu",
            "tf-nightly-gpu",
            "intel-tensorflow",
            "tensorflow-rocm",
            "tensorflow-macos",
        ]:
            try:
                A_      =							version.parse(importlib_metadata.version(package))
            except importlib_metadata.PackageNotFoundError:
                continue
            else:
                break
        else:
            A_      =							False
    if TF_AVAILABLE:
        if TF_VERSION.major < 2:
            logger.info(f'''TensorFlow found but with version {TF_VERSION}. `datasets` requires version 2 minimum.''')
            A_      =							False
        else:
            logger.info(f'''TensorFlow version {TF_VERSION} available.''')
else:
    logger.info("Disabling Tensorflow because USE_TORCH is set")


A_      =							"N/A"
A_      =							False

if USE_JAX in ENV_VARS_TRUE_AND_AUTO_VALUES:
    A_      =							importlib.util.find_spec("jax") is not None and importlib.util.find_spec("jaxlib") is not None
    if JAX_AVAILABLE:
        try:
            A_      =							version.parse(importlib_metadata.version("jax"))
            logger.info(f'''JAX version {JAX_VERSION} available.''')
        except importlib_metadata.PackageNotFoundError:
            pass
else:
    logger.info("Disabling JAX because USE_JAX is set to False")


A_      =							os.environ.get("USE_BEAM", "AUTO").upper()
A_      =							"N/A"
A_      =							False
if USE_BEAM in ENV_VARS_TRUE_AND_AUTO_VALUES:
    try:
        A_      =							version.parse(importlib_metadata.version("apache_beam"))
        A_      =							True
        logger.info(f'''Apache Beam version {BEAM_VERSION} available.''')
    except importlib_metadata.PackageNotFoundError:
        pass
else:
    logger.info("Disabling Apache Beam because USE_BEAM is set to False")


# Optional tools for data loading
A_      =							importlib.util.find_spec("sqlalchemy") is not None

# Optional tools for feature decoding
A_      =							importlib.util.find_spec("PIL") is not None
A_      =							importlib.util.find_spec("soundfile") is not None and version.parse(
    importlib.import_module("soundfile").__libsndfile_version__
) >= version.parse("1.0.31")
A_      =							importlib.util.find_spec("soundfile") is not None and version.parse(
    importlib.import_module("soundfile").__libsndfile_version__
) >= version.parse("1.1.0")

# Optional compression tools
A_      =							importlib.util.find_spec("rarfile") is not None
A_      =							importlib.util.find_spec("zstandard") is not None
A_      =							importlib.util.find_spec("lz4") is not None
A_      =							importlib.util.find_spec("py7zr") is not None


# Cache location
A_      =							"~/.cache"
A_      =							os.getenv("XDG_CACHE_HOME", DEFAULT_XDG_CACHE_HOME)
A_      =							os.path.join(XDG_CACHE_HOME, "huggingface")
A_      =							os.path.expanduser(os.getenv("HF_HOME", DEFAULT_HF_CACHE_HOME))

A_      =							os.path.join(HF_CACHE_HOME, "datasets")
A_      =							Path(os.getenv("HF_DATASETS_CACHE", DEFAULT_HF_DATASETS_CACHE))

A_      =							os.path.join(HF_CACHE_HOME, "metrics")
A_      =							Path(os.getenv("HF_METRICS_CACHE", DEFAULT_HF_METRICS_CACHE))

A_      =							os.path.join(HF_CACHE_HOME, "modules")
A_      =							Path(os.getenv("HF_MODULES_CACHE", DEFAULT_HF_MODULES_CACHE))

A_      =							"downloads"
A_      =							os.path.join(HF_DATASETS_CACHE, DOWNLOADED_DATASETS_DIR)
A_      =							Path(os.getenv("HF_DATASETS_DOWNLOADED_DATASETS_PATH", DEFAULT_DOWNLOADED_DATASETS_PATH))

A_      =							"extracted"
A_      =							os.path.join(DEFAULT_DOWNLOADED_DATASETS_PATH, EXTRACTED_DATASETS_DIR)
A_      =							Path(os.getenv("HF_DATASETS_EXTRACTED_DATASETS_PATH", DEFAULT_EXTRACTED_DATASETS_PATH))

# Download count for the website
A_      =							(
    os.environ.get("HF_UPDATE_DOWNLOAD_COUNTS", "AUTO").upper() in ENV_VARS_TRUE_AND_AUTO_VALUES
)

# Batch size constants. For more info, see:
# https://github.com/apache/arrow/blob/master/docs/source/cpp/arrays.rst#size-limitations-and-recommendations)
A_      =							10_00

# Size of the preloaded record batch in `Dataset.__iter__`
A_      =							10

# Pickling tables works only for small tables (<4GiB)
# For big tables, we write them on disk instead
A_      =							4 << 30

# Max shard size in bytes (e.g. to shard parquet datasets in push_to_hub or download_and_prepare)
A_      =							"500MB"

# Parquet configuration
A_      =							1_00
A_      =							1_00
A_      =							1_00

# Offline mode
A_      =							os.environ.get("HF_DATASETS_OFFLINE", "AUTO").upper() in ENV_VARS_TRUE_VALUES

# In-memory
A_      =							0  # Disabled
A_      =							float(os.environ.get("HF_DATASETS_IN_MEMORY_MAX_SIZE", DEFAULT_IN_MEMORY_MAX_SIZE))

# File names
A_      =							"dataset.arrow"
A_      =							"indices.arrow"
A_      =							"state.json"
A_      =							"dataset_info.json"
A_      =							"dataset_infos.json"
A_      =							"LICENSE"
A_      =							"metric_info.json"
A_      =							"dataset_dict.json"

A_      =							"datasets_modules"

A_      =							2_55

# Streaming
A_      =							20
A_      =							5

# Datasets without script
A_      =							2_00
A_      =							10
A_      =							2_00

# Progress bars
A_      =							0.05  # 20 progress updates per sec









'''simple docstring'''

import shutil
import tempfile
import unittest

import pytest

from transformers import WhisperTokenizer, is_speech_available
from transformers.testing_utils import require_sentencepiece, require_torch, require_torchaudio

from .test_feature_extraction_whisper import floats_list


if is_speech_available():
    from transformers import WhisperFeatureExtractor, WhisperProcessor


A_      =							5_03_58
A_      =							5_03_62





@require_torch
@require_torchaudio
@require_sentencepiece
class 							_snake_case	(    unittest.TestCase  ):
   def       __UpperCamelCase  (	self     : Any							):
     SCREAMING_SNAKE_CASE:int   =   "openai/whisper-small.en"
     SCREAMING_SNAKE_CASE:Optional[Any]   =   tempfile.mkdtemp()
   def       __UpperCamelCase  (	self     : Any       ,**SCREAMING_SNAKE_CASE__     : List[str]							):
     return WhisperTokenizer.from_pretrained(self.checkpoint       ,**SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : List[str]       ,**SCREAMING_SNAKE_CASE__     : Dict							):
     return WhisperFeatureExtractor.from_pretrained(self.checkpoint       ,**SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : str							):
     shutil.rmtree(self.tmpdirname							)
   def       __UpperCamelCase  (	self     : Any							):
     SCREAMING_SNAKE_CASE:Dict   =   self.get_tokenizer()
     SCREAMING_SNAKE_CASE:Optional[int]   =   self.get_feature_extractor()

     SCREAMING_SNAKE_CASE:Dict   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     processor.save_pretrained(self.tmpdirname							)
     SCREAMING_SNAKE_CASE:List[Any]   =   WhisperProcessor.from_pretrained(self.tmpdirname							)

     self.assertEqual(processor.tokenizer.get_vocab()       ,tokenizer.get_vocab()							)
     self.assertIsInstance(processor.tokenizer       ,SCREAMING_SNAKE_CASE__							)

     self.assertEqual(processor.feature_extractor.to_json_string()       ,feature_extractor.to_json_string()							)
     self.assertIsInstance(processor.feature_extractor       ,SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : Union[str, Any]							):
     SCREAMING_SNAKE_CASE:int   =   WhisperProcessor(tokenizer=self.get_tokenizer()       ,feature_extractor=self.get_feature_extractor()							)
     processor.save_pretrained(self.tmpdirname							)

     SCREAMING_SNAKE_CASE:List[str]   =   self.get_tokenizer(bos_token="(BOS)"       ,eos_token="(EOS)"							)
     SCREAMING_SNAKE_CASE:Tuple   =   self.get_feature_extractor(do_normalize=SCREAMING_SNAKE_CASE__       ,padding_value=1.0							)

     SCREAMING_SNAKE_CASE:Any   =   WhisperProcessor.from_pretrained(
         self.tmpdirname       ,bos_token="(BOS)"       ,eos_token="(EOS)"       ,do_normalize=SCREAMING_SNAKE_CASE__       ,padding_value=1.0							)

     self.assertEqual(processor.tokenizer.get_vocab()       ,tokenizer_add_kwargs.get_vocab()							)
     self.assertIsInstance(processor.tokenizer       ,SCREAMING_SNAKE_CASE__							)

     self.assertEqual(processor.feature_extractor.to_json_string()       ,feature_extractor_add_kwargs.to_json_string()							)
     self.assertIsInstance(processor.feature_extractor       ,SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : List[Any]							):
     SCREAMING_SNAKE_CASE:Any   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:List[Any]   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Optional[Any]   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:Any   =   floats_list((3, 1_000)							)

     SCREAMING_SNAKE_CASE:Any   =   feature_extractor(SCREAMING_SNAKE_CASE__       ,return_tensors="np"							)
     SCREAMING_SNAKE_CASE:List[str]   =   processor(SCREAMING_SNAKE_CASE__       ,return_tensors="np"							)

     for key in input_feat_extract.keys():
       self.assertAlmostEqual(input_feat_extract[key].sum()       ,input_processor[key].sum()       ,delta=1e-2							)
   def       __UpperCamelCase  (	self     : List[Any]							):
     SCREAMING_SNAKE_CASE:int   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:Tuple   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Union[str, Any]   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:str   =   "This is a test string"

     SCREAMING_SNAKE_CASE:int   =   processor(text=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:str   =   tokenizer(SCREAMING_SNAKE_CASE__							)

     for key in encoded_tok.keys():
       self.assertListEqual(encoded_tok[key]       ,encoded_processor[key]							)
   def       __UpperCamelCase  (	self     : str							):
     SCREAMING_SNAKE_CASE:List[Any]   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:List[str]   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Union[str, Any]   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     SCREAMING_SNAKE_CASE:Optional[Any]   =   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

     SCREAMING_SNAKE_CASE:Any   =   processor.batch_decode(SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:Union[str, Any]   =   tokenizer.batch_decode(SCREAMING_SNAKE_CASE__							)

     self.assertListEqual(SCREAMING_SNAKE_CASE__       ,SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : int							):
     SCREAMING_SNAKE_CASE:Tuple   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:Optional[int]   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Union[str, Any]   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)

     self.assertListEqual(
         processor.model_input_names       ,feature_extractor.model_input_names       ,msg="`processor` and `feature_extractor` model input names do not match"       ,)
   def       __UpperCamelCase  (	self     : int							):
     SCREAMING_SNAKE_CASE:Tuple   =   self.get_feature_extractor()
     SCREAMING_SNAKE_CASE:Any   =   self.get_tokenizer()

     SCREAMING_SNAKE_CASE:Any   =   WhisperProcessor(tokenizer=SCREAMING_SNAKE_CASE__       ,feature_extractor=SCREAMING_SNAKE_CASE__							)
     SCREAMING_SNAKE_CASE:Optional[int]   =   processor.get_decoder_prompt_ids(task="transcribe"       ,no_timestamps=SCREAMING_SNAKE_CASE__							)

     self.assertIsInstance(SCREAMING_SNAKE_CASE__       ,SCREAMING_SNAKE_CASE__							)
     for ids in forced_decoder_ids:
       self.assertIsInstance(SCREAMING_SNAKE_CASE__       ,(list, tuple)							)

     SCREAMING_SNAKE_CASE:Any   =   [TRANSCRIBE, NOTIMESTAMPS]
     self.assertListEqual([ids[-1] for ids in forced_decoder_ids]       ,SCREAMING_SNAKE_CASE__							)
   def       __UpperCamelCase  (	self     : Dict							):
     SCREAMING_SNAKE_CASE:Tuple   =   WhisperProcessor(tokenizer=self.get_tokenizer()       ,feature_extractor=self.get_feature_extractor()							)
     SCREAMING_SNAKE_CASE:Dict   =   processor.get_prompt_ids("Mr. Quilter"							)
     SCREAMING_SNAKE_CASE:int   =   processor.tokenizer.decode(SCREAMING_SNAKE_CASE__							)

     self.assertListEqual(prompt_ids.tolist()       ,[50_360, 1_770, 13, 2_264, 346, 353]							)
     self.assertEqual(SCREAMING_SNAKE_CASE__       ,"<|startofprev|> Mr. Quilter"							)
   def       __UpperCamelCase  (	self     : Union[str, Any]							):
     SCREAMING_SNAKE_CASE:Dict   =   WhisperProcessor(tokenizer=self.get_tokenizer()       ,feature_extractor=self.get_feature_extractor()							)
     SCREAMING_SNAKE_CASE:List[Any]   =   processor.get_prompt_ids(""							)
     SCREAMING_SNAKE_CASE:Tuple   =   processor.tokenizer.decode(SCREAMING_SNAKE_CASE__							)

     self.assertListEqual(prompt_ids.tolist()       ,[50_360, 220]							)
     self.assertEqual(SCREAMING_SNAKE_CASE__       ,"<|startofprev|> "							)







   def       __UpperCamelCase  (	self     : Union[str, Any]							):
     SCREAMING_SNAKE_CASE:Tuple   =   WhisperProcessor(tokenizer=self.get_tokenizer()       ,feature_extractor=self.get_feature_extractor()							)

     def _test_prompt_error_raised_helper(SCREAMING_SNAKE_CASE__     : int       ,SCREAMING_SNAKE_CASE__     : List[str]							):
       with pytest.raises(SCREAMING_SNAKE_CASE__							) as excinfo:
         processor.get_prompt_ids(SCREAMING_SNAKE_CASE__							)
       SCREAMING_SNAKE_CASE:List[Any]   =   F'''Encountered text in the prompt corresponding to disallowed special token: {special_token}.'''
       self.assertEqual(SCREAMING_SNAKE_CASE__       ,str(excinfo.value							)							)

     _test_prompt_error_raised_helper("<|startofprev|> test"       ,"<|startofprev|>"							)
     _test_prompt_error_raised_helper("test <|notimestamps|>"       ,"<|notimestamps|>"							)
     _test_prompt_error_raised_helper("test <|zh|> test <|transcribe|>"       ,"<|zh|>"							)






