'''simple docstring'''
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

  a=	True
except ImportError:
  a=	False


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







def 			_UpperCamelCase			(  _a						:						int      ):




    """simple docstring"""



    __UpperCamelCase						:						List[str]		     =	importlib.util.find_spec(_a      ) is not None
    if package_exists:
        try:
            __UpperCamelCase						:						Any		     =	importlib.metadata.metadata(_a      )
            return True
        except importlib.metadata.PackageNotFoundError:
            return False







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _torch_distributed_available







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    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 			_UpperCamelCase			(  ):




    """simple docstring"""



    return importlib.metadata.version('oneccl_bind_pt'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('transformer_engine'      )







@lru_cache
def 			_UpperCamelCase			(  _a						:						str=True      ):




    """simple docstring"""



    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						:						List[str]		     =	xm.xla_device()
            return True
        except RuntimeError:
            return False
    return _tpu_available







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('deepspeed'      )







def 			_UpperCamelCase			(  _a						:						List[str]=False      ):




    """simple docstring"""



    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 			_UpperCamelCase			(  ):




    """simple docstring"""



    __UpperCamelCase						:						Union[str, Any]		     =	_is_package_available('bitsandbytes'      )
    if package_exists:
        __UpperCamelCase						:						Union[str, Any]		     =	version.parse(importlib.metadata.version('bitsandbytes'      )      )
        return compare_versions(_a					,  '>='					,  '0.39.0'      )
    return False







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    __UpperCamelCase						:						str		     =	_is_package_available('bitsandbytes'      )
    if package_exists:
        __UpperCamelCase						:						Tuple		     =	version.parse(importlib.metadata.version('bitsandbytes'      )      )
        return compare_versions(_a					,  '>='					,  '0.37.2'      )
    return False







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('bitsandbytes'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



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







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('safetensors'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('transformers'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('datasets'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('aim'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('tensorboard'      ) or _is_package_available('tensorboardX'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('wandb'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('comet_ml'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('boto3'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    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'					,  _a      )
        return parse_flag_from_env('ACCELERATE_ENABLE_RICH'					,  _a      )
    return False







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('sagemaker'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('tqdm'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return _is_package_available('mlflow'      )







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    return is_torch_version('>='					,  '1.12'      ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()







def 			_UpperCamelCase			(  ):




    """simple docstring"""



    def get_major_and_minor_from_version(_a						:						Union[str, Any]      ):
        return str(version.parse(_a      ).major      ) + "." + str(version.parse(_a      ).minor      )

    __UpperCamelCase						:						int		     =	importlib.metadata.version('torch'      )
    if importlib.util.find_spec('intel_extension_for_pytorch'      ) is None:
        return False
    __UpperCamelCase						:						Optional[int]		     =	'N/A'
    try:
        __UpperCamelCase						:						int		     =	importlib.metadata.version('intel_extension_for_pytorch'      )
    except importlib.metadata.PackageNotFoundError:
        return False
    __UpperCamelCase						:						Union[str, Any]		     =	get_major_and_minor_from_version(_torch_version      )
    __UpperCamelCase						:						str		     =	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 			_UpperCamelCase			(  _a						:						Union[str, Any]=False      ):




    """simple docstring"""



    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						:						str		     =	torch.npu.device_count()
            return torch.npu.is_available()
        except RuntimeError:
            return False
    return hasattr(_a					,  'npu'      ) and torch.npu.is_available()







@lru_cache
def 			_UpperCamelCase			(  _a						:						Optional[Any]=False      ):




    """simple docstring"""



    if not parse_flag_from_env('ACCELERATE_USE_XPU'					,  default=_a      ):
        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						:						Tuple		     =	torch.xpu.device_count()
            return torch.xpu.is_available()
        except RuntimeError:
            return False
    return hasattr(_a					,  'xpu'      ) and torch.xpu.is_available()





'''simple docstring'''





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


A							     = get_tests_dir("fixtures/test_sentencepiece.model")



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



  '''simple docstring'''






  def 		_lowerCamelCase						(self	:			List[str]     ) ->							List[str]:
        snake_case_    							=				tempfile.mkdtemp()

        snake_case_    							=				["<s>", "<pad>", "</s>", "<unk>", "▁This", "▁is", "▁a", "▁t", "est"]
        snake_case_    							=				dict(zip(lowerCAmelCase_						,			range(len(lowerCAmelCase_     )     )     )     )
        snake_case_    							=				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"]     )

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

        snake_case_    							=				{
            "feature_size": 24,
            "num_mel_bins": 24,
            "padding_value": 0.0,
            "sampling_rate": 1_60_00,
            "return_attention_mask": False,
            "do_normalize": True,
        }
        save_json(lowerCAmelCase_						,			save_dir / FEATURE_EXTRACTOR_NAME     )

  def 		_lowerCamelCase						(self	:			Dict						,			**lowerCAmelCase_	:			Any     ) ->							Union[str, Any]:
        return SpeechaTextTokenizer.from_pretrained(self.tmpdirname						,			**lowerCAmelCase_     )

  def 		_lowerCamelCase						(self	:			List[str]						,			**lowerCAmelCase_	:			List[str]     ) ->							Any:
        return SpeechaTextFeatureExtractor.from_pretrained(self.tmpdirname						,			**lowerCAmelCase_     )

  def 		_lowerCamelCase						(self	:			int     ) ->							str:
        shutil.rmtree(self.tmpdirname     )

  def 		_lowerCamelCase						(self	:			List[Any]     ) ->							Optional[Any]:
        snake_case_    							=				self.get_tokenizer()
        snake_case_    							=				self.get_feature_extractor()

        snake_case_    							=				SpeechaTextProcessor(tokenizer=lowerCAmelCase_						,			feature_extractor=lowerCAmelCase_     )

        processor.save_pretrained(self.tmpdirname     )
        snake_case_    							=				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 		_lowerCamelCase						(self	:			str     ) ->							int:
        snake_case_    							=				SpeechaTextProcessor(
            tokenizer=self.get_tokenizer()						,			feature_extractor=self.get_feature_extractor()     )
        processor.save_pretrained(self.tmpdirname     )

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

        snake_case_    							=				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 		_lowerCamelCase						(self	:			Tuple     ) ->							Optional[Any]:
        snake_case_    							=				self.get_feature_extractor()
        snake_case_    							=				self.get_tokenizer()

        snake_case_    							=				SpeechaTextProcessor(tokenizer=lowerCAmelCase_						,			feature_extractor=lowerCAmelCase_     )

        snake_case_    							=				floats_list((3, 10_00)     )

        snake_case_    							=				feature_extractor(lowerCAmelCase_						,			return_tensors="np"     )
        snake_case_    							=				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 		_lowerCamelCase						(self	:			Union[str, Any]     ) ->							int:
        snake_case_    							=				self.get_feature_extractor()
        snake_case_    							=				self.get_tokenizer()

        snake_case_    							=				SpeechaTextProcessor(tokenizer=lowerCAmelCase_						,			feature_extractor=lowerCAmelCase_     )

        snake_case_    							=				"This is a test string"

        snake_case_    							=				processor(text=lowerCAmelCase_     )

        snake_case_    							=				tokenizer(lowerCAmelCase_     )

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

  def 		_lowerCamelCase						(self	:			Union[str, Any]     ) ->							int:
        snake_case_    							=				self.get_feature_extractor()
        snake_case_    							=				self.get_tokenizer()

        snake_case_    							=				SpeechaTextProcessor(tokenizer=lowerCAmelCase_						,			feature_extractor=lowerCAmelCase_     )

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

        snake_case_    							=				processor.batch_decode(lowerCAmelCase_     )
        snake_case_    							=				tokenizer.batch_decode(lowerCAmelCase_     )

        self.assertListEqual(lowerCAmelCase_						,			lowerCAmelCase_     )


  def 		_lowerCamelCase						(self	:			int     ) ->							int:
        snake_case_    							=				self.get_feature_extractor()
        snake_case_    							=				self.get_tokenizer()

        snake_case_    							=				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"						,			)

