




import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
    import torch_xla.core.xla_model as xm  # noqa: F401

    snake_case__ :						int     =			True
except ImportError:
    snake_case__ :						List[str]     =			False


# Cache this result has it's a C FFI call which can be pretty time-consuming
snake_case__ :						List[Any]     =			torch.distributed.is_available()


def 						lowercase		(     _lowerCAmelCase      ):
  UpperCAmelCase__    						=   importlib.util.find_spec(lowercase_      ) is not None
  if package_exists:
    try:
      UpperCAmelCase__    						=   importlib.metadata.metadata(lowercase_      )
      return True
    except importlib.metadata.PackageNotFoundError:
      return False


def 						lowercase		(     ):
  return _torch_distributed_available


def 						lowercase		(     ):







  try:
    pass
  except ImportError:
    print(
        """Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not"""
        """ detected. If you see \"ValueError: Invalid backend: 'ccl'\" error, please install Intel(R) oneCCL"""
        """ Bindings for PyTorch*."""      )
  return (
      importlib.util.find_spec("""torch_ccl"""      ) is not None
      or importlib.util.find_spec("""oneccl_bindings_for_pytorch"""      ) is not None
  )


def 						lowercase		(     ):
  return importlib.metadata.version("""oneccl_bind_pt"""      )


def 						lowercase		(     ):
  return _is_package_available("""transformer_engine"""      )


@lru_cache
def 						lowercase		(     _lowerCAmelCase=True      ):







  if torch.cuda.is_available():
    return False
  if _tpu_available and check_device:
    try:
      # Will raise a RuntimeError if no XLA configuration is found
      UpperCAmelCase__    						=   xm.xla_device()
      return True
    except RuntimeError:
      return False
  return _tpu_available


def 						lowercase		(     ):
  return _is_package_available("""deepspeed"""      )


def 						lowercase		(     _lowerCAmelCase=False      ):







  if is_tpu_available():
    return not ignore_tpu
  if torch.cuda.is_available():
    return torch.cuda.is_bfaa_supported()
  if is_npu_available():
    return False
  return True


def 						lowercase		(     ):
  UpperCAmelCase__    						=   _is_package_available("""bitsandbytes"""      )
  if package_exists:
    UpperCAmelCase__    						=   version.parse(importlib.metadata.version("""bitsandbytes"""      )      )
    return compare_versions(lowercase_   ,     """>="""   ,     """0.39.0"""      )
  return False


def 						lowercase		(     ):
  UpperCAmelCase__    						=   _is_package_available("""bitsandbytes"""      )
  if package_exists:
    UpperCAmelCase__    						=   version.parse(importlib.metadata.version("""bitsandbytes"""      )      )
    return compare_versions(lowercase_   ,     """>="""   ,     """0.37.2"""      )
  return False


def 						lowercase		(     ):
  return _is_package_available("""bitsandbytes"""      )


def 						lowercase		(     ):







  if strtobool(os.environ.get("""ACCELERATE_USE_MEGATRON_LM"""   ,     """False"""      )      ) == 1:
    UpperCAmelCase__    						=   importlib.util.find_spec("""megatron"""      ) is not None
    if package_exists:
      try:
        UpperCAmelCase__    						=   parse(importlib.metadata.version("""megatron-lm"""      )      )
        return compare_versions(lowercase_   ,     """>="""   ,     """2.2.0"""      )
      except Exception as e:
        warnings.warn(F'''Parse Megatron version failed. Exception:{e}'''      )
        return False


def 						lowercase		(     ):
  return _is_package_available("""safetensors"""      )


def 						lowercase		(     ):
  return _is_package_available("""transformers"""      )


def 						lowercase		(     ):
  return _is_package_available("""datasets"""      )


def 						lowercase		(     ):
  return _is_package_available("""aim"""      )


def 						lowercase		(     ):
  return _is_package_available("""tensorboard"""      ) or _is_package_available("""tensorboardX"""      )


def 						lowercase		(     ):
  return _is_package_available("""wandb"""      )


def 						lowercase		(     ):
  return _is_package_available("""comet_ml"""      )


def 						lowercase		(     ):
  return _is_package_available("""boto3"""      )


def 						lowercase		(     ):







  if _is_package_available("""rich"""      ):
    if "ACCELERATE_DISABLE_RICH" in os.environ:
      warnings.warn(
          """`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`."""      )
      return not parse_flag_from_env("""ACCELERATE_DISABLE_RICH"""   ,     lowercase_      )
    return parse_flag_from_env("""ACCELERATE_ENABLE_RICH"""   ,     lowercase_      )
  return False


def 						lowercase		(     ):
  return _is_package_available("""sagemaker"""      )


def 						lowercase		(     ):
  return _is_package_available("""tqdm"""      )


def 						lowercase		(     ):
  return _is_package_available("""mlflow"""      )


def 						lowercase		(     ):
  return is_torch_version(""">="""   ,     """1.12"""      ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()


def 						lowercase		(     ):







  def get_major_and_minor_from_version(_lowerCAmelCase      ):
    return str(version.parse(lowercase_      ).major      ) + "." + str(version.parse(lowercase_      ).minor      )

  UpperCAmelCase__    						=   importlib.metadata.version("""torch"""      )
  if importlib.util.find_spec("""intel_extension_for_pytorch"""      ) is None:
    return False
  UpperCAmelCase__    						=   """N/A"""
  try:
    UpperCAmelCase__    						=   importlib.metadata.version("""intel_extension_for_pytorch"""      )
  except importlib.metadata.PackageNotFoundError:
    return False
  UpperCAmelCase__    						=   get_major_and_minor_from_version(_torch_version      )
  UpperCAmelCase__    						=   get_major_and_minor_from_version(_ipex_version      )
  if torch_major_and_minor != ipex_major_and_minor:
    warnings.warn(
        F'''Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,'''
        F''' but PyTorch {_torch_version} is found. Please switch to the matching version and run again.'''      )
    return False
  return True


@lru_cache
def 						lowercase		(     _lowerCAmelCase=False      ):







  if importlib.util.find_spec("""torch"""      ) is None or importlib.util.find_spec("""torch_npu"""      ) is None:
    return False

  import torch
  import torch_npu  # noqa: F401

  if check_device:
    try:
      # Will raise a RuntimeError if no NPU is found
      UpperCAmelCase__    						=   torch.npu.device_count()
      return torch.npu.is_available()
    except RuntimeError:
      return False
  return hasattr(lowercase_   ,     """npu"""      ) and torch.npu.is_available()





@lru_cache
def 						lowercase		(     _lowerCAmelCase=False      ):







  if not parse_flag_from_env("""ACCELERATE_USE_XPU"""   ,     default=lowercase_      ):
    return False
  if is_ipex_available():
    import torch

    if is_torch_version("""<="""   ,     """1.12"""      ):
      return False
  else:
    return False

  import intel_extension_for_pytorch  # noqa: F401

  if check_device:
    try:
      # Will raise a RuntimeError if no XPU  is found
      UpperCAmelCase__    						=   torch.xpu.device_count()
      return torch.xpu.is_available()
    except RuntimeError:
      return False
  return hasattr(lowercase_   ,     """xpu"""      ) and torch.xpu.is_available()










import shutil
import tempfile
import unittest
from pathlib import Path
from shutil import copyfile

from transformers import SpeechaTextTokenizer, is_speech_available
from transformers.models.speech_to_text.tokenization_speech_to_text import VOCAB_FILES_NAMES, save_json
from transformers.testing_utils import get_tests_dir, require_sentencepiece, require_torch, require_torchaudio
from transformers.utils import FEATURE_EXTRACTOR_NAME

from .test_feature_extraction_speech_to_text import floats_list


if is_speech_available():
   from transformers import SpeechaTextFeatureExtractor, SpeechaTextProcessor


snake_case__ :						Union[str, Any]     =			get_tests_dir('''fixtures/test_sentencepiece.model''')





@require_torch
@require_torchaudio
@require_sentencepiece
class     snake_case  (	unittest.TestCase							):


   '''simple docstring'''







   def 	UpperCAmelCase       (					self     :	str			)					->str:
     '''simple docstring'''







     UpperCAmelCase__    						=   tempfile.mkdtemp()

     UpperCAmelCase__    						=   ["""<s>""", """<pad>""", """</s>""", """<unk>""", """▁This""", """▁is""", """▁a""", """▁t""", """est"""]
     UpperCAmelCase__    						=   dict(zip(lowerCamelCase_						,    range(len(lowerCamelCase_			)			)			)			)
     UpperCAmelCase__    						=   Path(self.tmpdirname			)
     save_json(lowerCamelCase_						,    save_dir / VOCAB_FILES_NAMES["""vocab_file"""]			)
     if not (save_dir / VOCAB_FILES_NAMES["spm_file"]).exists():
       copyfile(lowerCamelCase_						,    save_dir / VOCAB_FILES_NAMES["""spm_file"""]			)

     UpperCAmelCase__    						=   SpeechaTextTokenizer.from_pretrained(self.tmpdirname			)
     tokenizer.save_pretrained(self.tmpdirname			)

     UpperCAmelCase__    						=   {
         """feature_size""": 24,
         """num_mel_bins""": 24,
         """padding_value""": 0.0,
         """sampling_rate""": 1_6000,
         """return_attention_mask""": False,
         """do_normalize""": True,
     }
     save_json(lowerCamelCase_						,    save_dir / FEATURE_EXTRACTOR_NAME			)




   def 	UpperCAmelCase       (					self     :	Union[str, Any]						,    **lowerCamelCase_     :	Tuple			)					->Any:
     '''simple docstring'''







     return SpeechaTextTokenizer.from_pretrained(self.tmpdirname						,    **lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	Dict						,    **lowerCamelCase_     :	str			)					->Optional[Any]:
     '''simple docstring'''







     return SpeechaTextFeatureExtractor.from_pretrained(self.tmpdirname						,    **lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	List[str]			)					->Dict:
     '''simple docstring'''







     shutil.rmtree(self.tmpdirname			)




   def 	UpperCAmelCase       (					self     :	str			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_tokenizer()
     UpperCAmelCase__    						=   self.get_feature_extractor()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     processor.save_pretrained(self.tmpdirname			)
     UpperCAmelCase__    						=   SpeechaTextProcessor.from_pretrained(self.tmpdirname			)

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

     self.assertEqual(processor.feature_extractor.to_json_string()						,    feature_extractor.to_json_string()			)
     self.assertIsInstance(processor.feature_extractor						,    lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	List[str]			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   SpeechaTextProcessor(
         tokenizer=self.get_tokenizer()						,    feature_extractor=self.get_feature_extractor()			)
     processor.save_pretrained(self.tmpdirname			)

     UpperCAmelCase__    						=   self.get_tokenizer(bos_token="""(BOS)"""						,    eos_token="""(EOS)"""			)
     UpperCAmelCase__    						=   self.get_feature_extractor(do_normalize=lowerCamelCase_						,    padding_value=1.0			)

     UpperCAmelCase__    						=   SpeechaTextProcessor.from_pretrained(
         self.tmpdirname						,    bos_token="""(BOS)"""						,    eos_token="""(EOS)"""						,    do_normalize=lowerCamelCase_						,    padding_value=1.0			)

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

     self.assertEqual(processor.feature_extractor.to_json_string()						,    feature_extractor_add_kwargs.to_json_string()			)
     self.assertIsInstance(processor.feature_extractor						,    lowerCamelCase_			)




   def 	UpperCAmelCase       (					self     :	Union[str, Any]			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_feature_extractor()
     UpperCAmelCase__    						=   self.get_tokenizer()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     UpperCAmelCase__    						=   floats_list((3, 1000)			)

     UpperCAmelCase__    						=   feature_extractor(lowerCamelCase_						,    return_tensors="""np"""			)
     UpperCAmelCase__    						=   processor(lowerCamelCase_						,    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     :	str			)					->Union[str, Any]:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_feature_extractor()
     UpperCAmelCase__    						=   self.get_tokenizer()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     UpperCAmelCase__    						=   """This is a test string"""

     UpperCAmelCase__    						=   processor(text=lowerCamelCase_			)

     UpperCAmelCase__    						=   tokenizer(lowerCamelCase_			)

     for key in encoded_tok.keys():
       self.assertListEqual(encoded_tok[key]						,    encoded_processor[key]			)




   def 	UpperCAmelCase       (					self     :	Optional[Any]			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_feature_extractor()
     UpperCAmelCase__    						=   self.get_tokenizer()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     UpperCAmelCase__    						=   [[1, 4, 5, 8, 1, 0, 8], [3, 4, 3, 1, 1, 8, 9]]

     UpperCAmelCase__    						=   processor.batch_decode(lowerCamelCase_			)
     UpperCAmelCase__    						=   tokenizer.batch_decode(lowerCamelCase_			)

     self.assertListEqual(lowerCamelCase_						,    lowerCamelCase_			)







   def 	UpperCAmelCase       (					self     :	Dict			)					->Tuple:
     '''simple docstring'''







     UpperCAmelCase__    						=   self.get_feature_extractor()
     UpperCAmelCase__    						=   self.get_tokenizer()

     UpperCAmelCase__    						=   SpeechaTextProcessor(tokenizer=lowerCamelCase_						,    feature_extractor=lowerCamelCase_			)

     self.assertListEqual(
         processor.model_input_names						,    feature_extractor.model_input_names						,    msg="""`processor` and `feature_extractor` model input names do not match"""						,    )




