



'''simple docstring'''



import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment
def 					UpperCAmelCase   (						A  :				Any					,		A  :				Optional[int]=()					,		A  :				List[str]=None					,		A  :				Optional[int]="no"					,		A  :				Union[str, Any]="29500"				):
							SCREAMING_SNAKE_CASE					:      Any     					=			False
							SCREAMING_SNAKE_CASE					:      Any     					=			False
							if any(key.startswith('''KAGGLE'''				) for key in os.environ.keys()				):
														SCREAMING_SNAKE_CASE					:      List[Any]     					=			True
							elif "IPython" in sys.modules:
														SCREAMING_SNAKE_CASE					:      List[str]     					=			'''google.colab''' in str(sys.modules['''IPython'''].get_ipython()				)

							try:
														SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			PrecisionType(mixed_precision.lower()				)
							except ValueError:
														raise ValueError(
														    F"""Unknown mixed_precision mode: {args.mixed_precision.lower()}. Choose between {PrecisionType.list()}."""				)

							if (in_colab or in_kaggle) and (os.environ.get('''TPU_NAME'''					,		A				) is not None):
														# TPU launch
														import torch_xla.distributed.xla_multiprocessing as xmp

														if len(AcceleratorState._shared_state				) > 0:
																					raise ValueError(
																					    '''To train on TPU in Colab or Kaggle Kernel, the `Accelerator` should only be initialized inside '''
																					    '''your training function. Restart your notebook and make sure no cells initializes an '''
																					    '''`Accelerator`.'''				)
														if num_processes is None:
																					SCREAMING_SNAKE_CASE					:      Dict     					=			8

														SCREAMING_SNAKE_CASE					:      List[str]     					=			PrepareForLaunch(A					,		distributed_type='''TPU'''				)
														print(F"""Launching a training on {num_processes} TPU cores."""				)
														xmp.spawn(A					,		args=A					,		nprocs=A					,		start_method='''fork'''				)
							elif in_colab:
														# No need for a distributed launch otherwise as it's either CPU or one GPU.
														if torch.cuda.is_available():
																					print('''Launching training on one GPU.'''				)
														else:
																					print('''Launching training on one CPU.'''				)
														function(*A				)
							else:
														if num_processes is None:
																					raise ValueError(
																					    '''You have to specify the number of GPUs you would like to use, add `num_processes=...` to your call.'''				)

														if num_processes > 1:
																					# Multi-GPU launch
																					from torch.multiprocessing import start_processes
																					from torch.multiprocessing.spawn import ProcessRaisedException

																					if len(AcceleratorState._shared_state				) > 0:
																												raise ValueError(
																												    '''To launch a multi-GPU training from your notebook, the `Accelerator` should only be initialized '''
																												    '''inside your training function. Restart your notebook and make sure no cells initializes an '''
																												    '''`Accelerator`.'''				)

																					if torch.cuda.is_initialized():
																												raise ValueError(
																												    '''To launch a multi-GPU training from your notebook, you need to avoid running any instruction '''
																												    '''using `torch.cuda` in any cell. Restart your notebook and make sure no cells use any CUDA '''
																												    '''function.'''				)

																					# torch.distributed will expect a few environment variable to be here. We set the ones common to each
																					# process here (the other ones will be set be the launcher).
																					with patch_environment(
																					    world_size=A					,		master_addr='''127.0.01'''					,		master_port=A					,		mixed_precision=A				):
																												SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			PrepareForLaunch(A					,		distributed_type='''MULTI_GPU'''				)
																												print(F"""Launching training on {num_processes} GPUs."""				)
																												try:
																																			start_processes(A					,		args=A					,		nprocs=A					,		start_method='''fork'''				)
																												except ProcessRaisedException as e:
																																			if "Cannot re-initialize CUDA in forked subprocess" in e.args[0]:
																																										raise RuntimeError(
																																										    '''CUDA has been initialized before the `notebook_launcher` could create a forked subprocess. '''
																																										    '''This likely stems from an outside import causing issues once the `notebook_launcher()` is called. '''
																																										    '''Please review your imports and test them when running the `notebook_launcher()` to identify '''
																																										    '''which one is problematic.'''				) from e

														else:
																					# No need for a distributed launch otherwise as it's either CPU, GPU or MPS.
																					if is_mps_available():
																												SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			'''1'''
																												print('''Launching training on MPS.'''				)
																					elif torch.cuda.is_available():
																												print('''Launching training on one GPU.'''				)
																					else:
																												print('''Launching training on CPU.'''				)
																					function(*A				)






def 					UpperCAmelCase   (						A  :				Optional[Any]					,		A  :				Optional[Any]=()					,		A  :				str=2				):
							from torch.multiprocessing import start_processes

							with tempfile.NamedTemporaryFile() as tmp_file:
														# torch.distributed will expect a few environment variable to be here. We set the ones common to each
														# process here (the other ones will be set be the launcher).
														with patch_environment(
														    world_size=A					,		master_addr='''127.0.01'''					,		master_port='''29500'''					,		accelerate_mixed_precision='''no'''					,		accelerate_debug_rdv_file=tmp_file.name					,		accelerate_use_cpu='''yes'''					,		):
																					SCREAMING_SNAKE_CASE					:      Optional[int]     					=			PrepareForLaunch(A					,		debug=A				)
																					start_processes(A					,		args=A					,		nprocs=A					,		start_method='''fork'''				)









'''simple docstring'''



import shutil
import tempfile
import unittest

from transformers import ClapFeatureExtractor, ClapProcessor, RobertaTokenizer, RobertaTokenizerFast
from transformers.testing_utils import require_sentencepiece, require_torchaudio

from .test_feature_extraction_clap import floats_list
@require_torchaudio
@require_sentencepiece
class 	lowerCamelCase_							(					unittest.TestCase			):







						def 					__lowercase    (  self   :      Union[str, Any]				):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE					:      int     					=			'''laion/clap-htsat-unfused'''
													SCREAMING_SNAKE_CASE					:      Tuple     					=			tempfile.mkdtemp()







						def 					__lowercase    (  self   :      Tuple				,    **lowerCAmelCase__   :      Dict				):

													"""simple docstring"""
													return RobertaTokenizer.from_pretrained(self.checkpoint				,    **lowerCAmelCase__				)







						def 					__lowercase    (  self   :      int				,    **lowerCAmelCase__   :      str				):

													"""simple docstring"""
													return ClapFeatureExtractor.from_pretrained(self.checkpoint				,    **lowerCAmelCase__				)







						def 					__lowercase    (  self   :      Union[str, Any]				):

													"""simple docstring"""
													shutil.rmtree(self.tmpdirname				)







						def 					__lowercase    (  self   :      Dict				):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE					:      List[str]     					=			self.get_tokenizer()
													SCREAMING_SNAKE_CASE					:      Optional[int]     					=			self.get_feature_extractor()

													SCREAMING_SNAKE_CASE					:      List[str]     					=			ClapProcessor(tokenizer=lowerCAmelCase__				,    feature_extractor=lowerCAmelCase__				)

													processor.save_pretrained(self.tmpdirname				)
													SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			ClapProcessor.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 					__lowercase    (  self   :      Union[str, Any]				):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE					:      Dict     					=			ClapProcessor(tokenizer=self.get_tokenizer()				,    feature_extractor=self.get_feature_extractor()				)
													processor.save_pretrained(self.tmpdirname				)

													SCREAMING_SNAKE_CASE					:      List[Any]     					=			self.get_tokenizer(bos_token='''(BOS)'''				,    eos_token='''(EOS)'''				)
													SCREAMING_SNAKE_CASE					:      List[Any]     					=			self.get_feature_extractor(do_normalize=lowerCAmelCase__				,    padding_value=1.0				)

													SCREAMING_SNAKE_CASE					:      str     					=			ClapProcessor.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 					__lowercase    (  self   :      Dict				):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE					:      List[Any]     					=			self.get_feature_extractor()
													SCREAMING_SNAKE_CASE					:      Tuple     					=			self.get_tokenizer()

													SCREAMING_SNAKE_CASE					:      Tuple     					=			ClapProcessor(tokenizer=lowerCAmelCase__				,    feature_extractor=lowerCAmelCase__				)

													SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			floats_list((3, 10_00)				)

													SCREAMING_SNAKE_CASE					:      Any     					=			feature_extractor(lowerCAmelCase__				,    return_tensors='''np'''				)
													SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			processor(audios=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 					__lowercase    (  self   :      Any				):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE					:      Tuple     					=			self.get_feature_extractor()
													SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			self.get_tokenizer()

													SCREAMING_SNAKE_CASE					:      Optional[int]     					=			ClapProcessor(tokenizer=lowerCAmelCase__				,    feature_extractor=lowerCAmelCase__				)

													SCREAMING_SNAKE_CASE					:      int     					=			'''This is a test string'''

													SCREAMING_SNAKE_CASE					:      int     					=			processor(text=lowerCAmelCase__				)

													SCREAMING_SNAKE_CASE					:      Optional[int]     					=			tokenizer(lowerCAmelCase__				)

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







						def 					__lowercase    (  self   :      Optional[Any]				):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE					:      str     					=			self.get_feature_extractor()
													SCREAMING_SNAKE_CASE					:      List[str]     					=			self.get_tokenizer()

													SCREAMING_SNAKE_CASE					:      Any     					=			ClapProcessor(tokenizer=lowerCAmelCase__				,    feature_extractor=lowerCAmelCase__				)

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

													SCREAMING_SNAKE_CASE					:      Tuple     					=			processor.batch_decode(lowerCAmelCase__				)
													SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			tokenizer.batch_decode(lowerCAmelCase__				)

													self.assertListEqual(lowerCAmelCase__				,    lowerCAmelCase__				)







						def 					__lowercase    (  self   :      Optional[Any]				):

													"""simple docstring"""
													SCREAMING_SNAKE_CASE					:      Union[str, Any]     					=			self.get_feature_extractor()
													SCREAMING_SNAKE_CASE					:      str     					=			self.get_tokenizer()

													SCREAMING_SNAKE_CASE					:      Optional[Any]     					=			ClapProcessor(tokenizer=lowerCAmelCase__				,    feature_extractor=lowerCAmelCase__				)

													self.assertListEqual(
													    processor.model_input_names[2:]				,    feature_extractor.model_input_names				,    msg='''`processor` and `feature_extractor` model input names do not match'''				,    )




