
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

    __lowerCAmelCase      :					int   	=       True
except ImportError:
    __lowerCAmelCase      :					Dict   	=       False


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

def a__   (	A_ ):






  '''simple docstring'''





  __magic_name__    					=	importlib.util.find_spec(A_ ) is not None
  if package_exists:
    try:
      __magic_name__    					=	importlib.metadata.metadata(A_ )
      return True
    except importlib.metadata.PackageNotFoundError:
      return False

def a__   (	):






  '''simple docstring'''





  return _torch_distributed_available

def a__   (	):






  '''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 a__   (	):






  '''simple docstring'''





  return importlib.metadata.version("""oneccl_bind_pt""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""transformer_engine""" )

@lru_cache
def a__   (	A_=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
      __magic_name__    					=	xm.xla_device()
      return True
    except RuntimeError:
      return False
  return _tpu_available

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""deepspeed""" )

def a__   (	A_=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 a__   (	):






  '''simple docstring'''





  __magic_name__    					=	_is_package_available("""bitsandbytes""" )
  if package_exists:
    __magic_name__    					=	version.parse(importlib.metadata.version("""bitsandbytes""" ) )
    return compare_versions(A_,					""">=""",					"""0.39.0""" )
  return False

def a__   (	):






  '''simple docstring'''





  __magic_name__    					=	_is_package_available("""bitsandbytes""" )
  if package_exists:
    __magic_name__    					=	version.parse(importlib.metadata.version("""bitsandbytes""" ) )
    return compare_versions(A_,					""">=""",					"""0.37.2""" )
  return False

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""bitsandbytes""" )

def a__   (	):






  '''simple docstring'''





  if strtobool(os.environ.get("""ACCELERATE_USE_MEGATRON_LM""",					"""False""" ) ) == 1:
    __magic_name__    					=	importlib.util.find_spec("""megatron""" ) is not None
    if package_exists:
      try:
        __magic_name__    					=	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 a__   (	):






  '''simple docstring'''





  return _is_package_available("""safetensors""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""transformers""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""datasets""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""aim""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""tensorboard""" ) or _is_package_available("""tensorboardX""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""wandb""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""comet_ml""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""boto3""" )

def a__   (	):






  '''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 a__   (	):






  '''simple docstring'''





  return _is_package_available("""sagemaker""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""tqdm""" )

def a__   (	):






  '''simple docstring'''





  return _is_package_available("""mlflow""" )

def a__   (	):






  '''simple docstring'''





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

def a__   (	):






  '''simple docstring'''





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

  __magic_name__    					=	importlib.metadata.version("""torch""" )
  if importlib.util.find_spec("""intel_extension_for_pytorch""" ) is None:
    return False
  __magic_name__    					=	"""N/A"""
  try:
    __magic_name__    					=	importlib.metadata.version("""intel_extension_for_pytorch""" )
  except importlib.metadata.PackageNotFoundError:
    return False
  __magic_name__    					=	get_major_and_minor_from_version(_torch_version )
  __magic_name__    					=	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 a__   (	A_=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
      __magic_name__    					=	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 a__   (	A_=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
      __magic_name__    					=	torch.xpu.device_count()
      return torch.xpu.is_available()
    except RuntimeError:
      return False
  return hasattr(A_,					"""xpu""" ) and torch.xpu.is_available()



import random
import shutil
import tempfile
import unittest

import numpy as np

from transformers import TaTokenizer, TaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torch
from transformers.utils.import_utils import is_speech_available, is_torch_available


if is_torch_available():
    pass

if is_speech_available():
    from transformers import EncodecFeatureExtractor, MusicgenProcessor


__lowerCAmelCase      :					Dict   	=       random.Random()

def a__   (	A_,					A_=1.0,					A_=None,					A_=None ):






  '''simple docstring'''





  if rng is None:
    __magic_name__    					=	global_rng

  __magic_name__    					=	[]
  for batch_idx in range(shape[0] ):
    values.append([] )
    for _ in range(shape[1] ):
      values[-1].append(rng.random() * scale )

  return values





@require_torch
@require_sentencepiece
class 	UpperCAmelCase_      (			unittest.TestCase					):


  '''simple docstring'''







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







    """simple docstring"""






    __magic_name__    					=	"""facebook/musicgen-small"""
    __magic_name__    					=	tempfile.mkdtemp()
  def 							_lowercase							(			self      :			Tuple      ,							**UpperCamelCase__      :			str   )      ->     Any:







    """simple docstring"""






    return TaTokenizer.from_pretrained(self.checkpoint      ,							**UpperCamelCase__   )
  def 							_lowercase							(			self      :			Optional[int]      ,							**UpperCamelCase__      :			List[str]   )      ->     List[str]:







    """simple docstring"""






    return EncodecFeatureExtractor.from_pretrained(self.checkpoint      ,							**UpperCamelCase__   )
  def 							_lowercase							(			self      :			Union[str, Any]   )      ->     List[Any]:







    """simple docstring"""






    shutil.rmtree(self.tmpdirname   )
  def 							_lowercase							(			self      :			Optional[Any]   )      ->     Optional[int]:







    """simple docstring"""






    __magic_name__    					=	self.get_tokenizer()
    __magic_name__    					=	self.get_feature_extractor()

    __magic_name__    					=	MusicgenProcessor(tokenizer=UpperCamelCase__      ,							feature_extractor=UpperCamelCase__   )

    processor.save_pretrained(self.tmpdirname   )
    __magic_name__    					=	MusicgenProcessor.from_pretrained(self.tmpdirname   )

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

    self.assertEqual(processor.feature_extractor.to_json_string()      ,							feature_extractor.to_json_string()   )
    self.assertIsInstance(processor.feature_extractor      ,							UpperCamelCase__   )
  def 							_lowercase							(			self      :			Optional[Any]   )      ->     List[str]:







    """simple docstring"""






    __magic_name__    					=	MusicgenProcessor(tokenizer=self.get_tokenizer()      ,							feature_extractor=self.get_feature_extractor()   )
    processor.save_pretrained(self.tmpdirname   )

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

    __magic_name__    					=	MusicgenProcessor.from_pretrained(
        self.tmpdirname      ,							bos_token="""(BOS)"""      ,							eos_token="""(EOS)"""      ,							do_normalize=UpperCamelCase__      ,							padding_value=1.0   )

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

    self.assertEqual(processor.feature_extractor.to_json_string()      ,							feature_extractor_add_kwargs.to_json_string()   )
    self.assertIsInstance(processor.feature_extractor      ,							UpperCamelCase__   )
  def 							_lowercase							(			self      :			Dict   )      ->     Optional[Any]:







    """simple docstring"""






    __magic_name__    					=	self.get_feature_extractor()
    __magic_name__    					=	self.get_tokenizer()

    __magic_name__    					=	MusicgenProcessor(tokenizer=UpperCamelCase__      ,							feature_extractor=UpperCamelCase__   )

    __magic_name__    					=	floats_list((3, 1000)   )

    __magic_name__    					=	feature_extractor(UpperCamelCase__      ,							return_tensors="""np"""   )
    __magic_name__    					=	processor(UpperCamelCase__      ,							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 							_lowercase							(			self      :			Optional[Any]   )      ->     str:







    """simple docstring"""






    __magic_name__    					=	self.get_feature_extractor()
    __magic_name__    					=	self.get_tokenizer()

    __magic_name__    					=	MusicgenProcessor(tokenizer=UpperCamelCase__      ,							feature_extractor=UpperCamelCase__   )

    __magic_name__    					=	"""This is a test string"""

    __magic_name__    					=	processor(text=UpperCamelCase__   )

    __magic_name__    					=	tokenizer(UpperCamelCase__   )

    for key in encoded_tok.keys():
      self.assertListEqual(encoded_tok[key]      ,							encoded_processor[key]   )
  def 							_lowercase							(			self      :			Dict   )      ->     int:







    """simple docstring"""






    __magic_name__    					=	self.get_feature_extractor()
    __magic_name__    					=	self.get_tokenizer()

    __magic_name__    					=	MusicgenProcessor(tokenizer=UpperCamelCase__      ,							feature_extractor=UpperCamelCase__   )

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

    __magic_name__    					=	processor.batch_decode(sequences=UpperCamelCase__   )
    __magic_name__    					=	tokenizer.batch_decode(UpperCamelCase__   )

    self.assertListEqual(UpperCamelCase__      ,							UpperCamelCase__   )
  def 							_lowercase							(			self      :			Tuple   )      ->     int:







    """simple docstring"""






    __magic_name__    					=	self.get_feature_extractor()
    __magic_name__    					=	self.get_tokenizer()

    __magic_name__    					=	MusicgenProcessor(tokenizer=UpperCamelCase__      ,							feature_extractor=UpperCamelCase__   )

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





  def 							_lowercase							(			self      :			List[str]   )      ->     int:







    """simple docstring"""






    __magic_name__    					=	self.get_feature_extractor(padding_side="""left"""   )
    __magic_name__    					=	self.get_tokenizer()

    __magic_name__    					=	MusicgenProcessor(tokenizer=UpperCamelCase__      ,							feature_extractor=UpperCamelCase__   )

    __magic_name__    					=	[floats_list((1, x)   )[0] for x in range(5      ,							20      ,							5   )]
    __magic_name__    					=	processor(UpperCamelCase__   ).padding_mask

    __magic_name__    					=	np.asarray(floats_list((3, 20)   )   )[:, None, :]
    __magic_name__    					=	processor.batch_decode(UpperCamelCase__      ,							padding_mask=UpperCamelCase__   )

    self.assertIsInstance(UpperCamelCase__      ,							UpperCamelCase__   )

    for audio in decoded_audios:
      self.assertIsInstance(UpperCamelCase__      ,							np.ndarray   )

    self.assertTrue(decoded_audios[0].shape == (1, 10)   )
    self.assertTrue(decoded_audios[1].shape == (1, 15)   )
    self.assertTrue(decoded_audios[2].shape == (1, 20)   )

