


import shutil
import tempfile
import unittest

import numpy as np
import pytest

from transformers import is_speech_available, is_vision_available
from transformers.testing_utils import require_torch


if is_vision_available():
							from transformers import TvltImageProcessor

if is_speech_available():
							from transformers import TvltFeatureExtractor

from transformers import TvltProcessor

@require_torch
class 						A__							(		unittest.TestCase  ):




						def 							__UpperCamelCase				(       self					:      Any  )							->     Optional[int]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE			  ='''ZinengTang/tvlt-base'''
										_SCREAMING_SNAKE_CASE			  =tempfile.mkdtemp()




						def 							__UpperCamelCase				(       self					:      int				,    **_a					:      List[Any]  )							->     str:


										"""simple docstring"""





										return TvltImageProcessor.from_pretrained(self.checkpoint				,    **_a  )




						def 							__UpperCamelCase				(       self					:      Optional[int]				,    **_a					:      Union[str, Any]  )							->     str:


										"""simple docstring"""





										return TvltFeatureExtractor.from_pretrained(self.checkpoint				,    **_a  )




						def 							__UpperCamelCase				(       self					:      Optional[int]  )							->     Optional[Any]:


										"""simple docstring"""





										shutil.rmtree(self.tmpdirname  )




						def 							__UpperCamelCase				(       self					:      int  )							->     List[Any]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE			  =self.get_image_processor()
										_SCREAMING_SNAKE_CASE			  =self.get_feature_extractor()

										_SCREAMING_SNAKE_CASE			  =TvltProcessor(image_processor=_a				,    feature_extractor=_a  )
										processor.save_pretrained(self.tmpdirname  )
										_SCREAMING_SNAKE_CASE			  =TvltProcessor.from_pretrained(self.tmpdirname  )

										self.assertIsInstance(processor.feature_extractor				,    _a  )
										self.assertIsInstance(processor.image_processor				,    _a  )




						def 							__UpperCamelCase				(       self					:      Tuple  )							->     Any:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE			  =self.get_image_processor()
										_SCREAMING_SNAKE_CASE			  =self.get_feature_extractor()

										_SCREAMING_SNAKE_CASE			  =TvltProcessor(image_processor=_a				,    feature_extractor=_a  )

										_SCREAMING_SNAKE_CASE			  =np.ones([1_2000]  )

										_SCREAMING_SNAKE_CASE			  =feature_extractor(_a				,    return_tensors='''np'''  )
										_SCREAMING_SNAKE_CASE			  =processor(audio=_a				,    return_tensors='''np'''  )

										for key in audio_dict.keys():
														self.assertAlmostEqual(audio_dict[key].sum()				,    input_processor[key].sum()				,    delta=1E-2  )




						def 							__UpperCamelCase				(       self					:      Optional[int]  )							->     List[Any]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE			  =self.get_image_processor()
										_SCREAMING_SNAKE_CASE			  =self.get_feature_extractor()

										_SCREAMING_SNAKE_CASE			  =TvltProcessor(image_processor=_a				,    feature_extractor=_a  )

										_SCREAMING_SNAKE_CASE			  =np.ones([3, 224, 224]  )

										_SCREAMING_SNAKE_CASE			  =image_processor(_a				,    return_tensors='''np'''  )
										_SCREAMING_SNAKE_CASE			  =processor(images=_a				,    return_tensors='''np'''  )

										for key in image_dict.keys():
														self.assertAlmostEqual(image_dict[key].sum()				,    input_processor[key].sum()				,    delta=1E-2  )




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


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE			  =self.get_image_processor()
										_SCREAMING_SNAKE_CASE			  =self.get_feature_extractor()

										_SCREAMING_SNAKE_CASE			  =TvltProcessor(image_processor=_a				,    feature_extractor=_a  )

										_SCREAMING_SNAKE_CASE			  =np.ones([1_2000]  )
										_SCREAMING_SNAKE_CASE			  =np.ones([3, 224, 224]  )

										_SCREAMING_SNAKE_CASE			  =processor(audio=_a				,    images=_a  )

										self.assertListEqual(list(inputs.keys()  )				,    ['''audio_values''', '''audio_mask''', '''pixel_values''', '''pixel_mask''']  )

										# test if it raises when no input is passed
										with pytest.raises(_a  ):
														processor()




						def 							__UpperCamelCase				(       self					:      Dict  )							->     List[Any]:


										"""simple docstring"""





										_SCREAMING_SNAKE_CASE			  =self.get_image_processor()
										_SCREAMING_SNAKE_CASE			  =self.get_feature_extractor()

										_SCREAMING_SNAKE_CASE			  =TvltProcessor(image_processor=_a				,    feature_extractor=_a  )

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



import os
import sys
import tempfile

import torch

from .state import AcceleratorState
from .utils import PrecisionType, PrepareForLaunch, is_mps_available, patch_environment







def 						lowerCamelCase(   a__	,a__=()	,a__=None	,a__="no"	,a__="29500"):
				_SCREAMING_SNAKE_CASE			  =False
				_SCREAMING_SNAKE_CASE			  =False
				if any(key.startswith('''KAGGLE''') for key in os.environ.keys()):
								_SCREAMING_SNAKE_CASE			  =True
				elif "IPython" in sys.modules:
								_SCREAMING_SNAKE_CASE			  ='''google.colab''' in str(sys.modules['''IPython'''].get_ipython())

				try:
								_SCREAMING_SNAKE_CASE			  =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			  =8

								_SCREAMING_SNAKE_CASE			  =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			  =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			  ='''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 						lowerCamelCase(   a__	,a__=()	,a__=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			  =PrepareForLaunch(a__	,debug=a__)
												start_processes(a__	,args=a__	,nprocs=a__	,start_method='''fork''')