





import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


__UpperCamelCase:	Any = logging.get_logger(__name__)


def       SCREAMING_SNAKE_CASE__  (       )   ->							Any:




		'''simple docstring'''
		# Get the sagemaker specific mp parameters from smp_options variable.
		lowercase__					:      str      =		os.getenv('SM_HP_MP_PARAMETERS'   ,    '{}'			)
		try:
				# Parse it and check the field "partitions" is included, it is required for model parallel.
				lowercase__					:      str      =		json.loads(_lowercase			)
				if "partitions" not in smp_options:
						return False
		except json.JSONDecodeError:
				return False

		# Get the sagemaker specific framework parameters from mpi_options variable.
		lowercase__					:      Dict      =		os.getenv('SM_FRAMEWORK_PARAMS'   ,    '{}'			)
		try:
				# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
				lowercase__					:      Union[str, Any]      =		json.loads(_lowercase			)
				if not mpi_options.get('sagemaker_mpi_enabled'   ,    _lowercase			):
						return False
		except json.JSONDecodeError:
				return False
		# Lastly, check if the `smdistributed` module is present.
		return importlib.util.find_spec('smdistributed'			) is not None


if is_sagemaker_model_parallel_available():
			import smdistributed.modelparallel.torch as smp

			smp.init()





@dataclass
class       __lowerCAmelCase					(       _UpperCamelCase      ):


	'''simple docstring'''





	_A													=				field(
	    default=""   ,    metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}   ,    )

	def    snake_case__(						self:						int			):
			super().__post_init__()
			warnings.warn(
			    '`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use '
			    '`TrainingArguments` instead.',					lowerCamelCase_,					)

	@cached_property
	def    snake_case__(						self:						Union[str, Any]			):
			logger.info('PyTorch: setting up devices'			)
			if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
					logger.warning(
					    'torch.distributed process group is initialized, but local_rank == -1. '
					    'In order to use Torch DDP, launch your script with `python -m torch.distributed.launch'			)
			if self.no_cuda:
					lowercase__					:      Tuple      =		torch.device('cpu'			)
					lowercase__					:      Optional[int]      =		0
			elif is_sagemaker_model_parallel_available():
					lowercase__					:      Dict      =		smp.local_rank()
					lowercase__					:      List[str]      =		torch.device('cuda',					lowerCamelCase_			)
					lowercase__					:      Optional[Any]      =		1
			elif is_sagemaker_dp_enabled():
					import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

					torch.distributed.init_process_group(backend='smddp',					timeout=self.ddp_timeout_delta			)
					lowercase__					:      Dict      =		int(os.getenv('SMDATAPARALLEL_LOCAL_RANK'			)			)
					lowercase__					:      Optional[Any]      =		torch.device('cuda',					self.local_rank			)
					lowercase__					:      List[Any]      =		1
			elif self.local_rank == -1:
					# if n_gpu is > 1 we'll use nn.DataParallel.
					# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
					# Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
					# trigger an error that a device index is missing. Index 0 takes into account the
					# GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
					# will use the first GPU in that env, i.e. GPU#1
					lowercase__					:      int      =		torch.device('cuda:0' if torch.cuda.is_available() else 'cpu'			)
					# Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
					# the default value.
					lowercase__					:      int      =		torch.cuda.device_count()
			else:
					# Here, we'll use torch.distributed.
					# Initializes the distributed backend which will take care of synchronizing nodes/GPUs
					if not torch.distributed.is_initialized():
							torch.distributed.init_process_group(backend='nccl',					timeout=self.ddp_timeout_delta			)
					lowercase__					:      Union[str, Any]      =		torch.device('cuda',					self.local_rank			)
					lowercase__					:      Dict      =		1

			if device.type == "cuda":
					torch.cuda.set_device(lowerCamelCase_			)

			return device

	@property
	def    snake_case__(						self:						Any			):
			if is_sagemaker_model_parallel_available():
					return smp.dp_size()

			return super().world_size

	@property
	def    snake_case__(						self:						Tuple			):
			return not is_sagemaker_model_parallel_available()





	@property
	def    snake_case__(						self:						List[Any]			):
			return False









import unittest

import numpy as np

from transformers import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING, TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
from transformers.pipelines import AudioClassificationPipeline, pipeline
from transformers.testing_utils import (
    is_pipeline_test,
    nested_simplify,
    require_tf,
    require_torch,
    require_torchaudio,
    slow,
)

from .test_pipelines_common import ANY


@is_pipeline_test
class       __lowerCAmelCase					(       unittest.TestCase      ):


	'''simple docstring'''





	_A													=				MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING
	_A													=				TF_MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

	def    snake_case__(						self:						Optional[int],					lowerCamelCase_:						Tuple,					lowerCamelCase_:						List[Any],					lowerCamelCase_:						Optional[Any]			):
			lowercase__					:      int      =		AudioClassificationPipeline(model=lowerCamelCase_,					feature_extractor=lowerCamelCase_			)

			# test with a raw waveform
			lowercase__					:      Dict      =		np.zeros((34000,)			)
			lowercase__					:      Optional[int]      =		np.zeros((14000,)			)
			return audio_classifier, [audioa, audio]

	def    snake_case__(						self:						Optional[int],					lowerCamelCase_:						Tuple,					lowerCamelCase_:						str			):
			lowercase__   ,				lowercase__					:      List[str]      =		examples
			lowercase__					:      int      =		audio_classifier(lowerCamelCase_			)
			# by default a model is initialized with num_labels=2
			self.assertEqual(
			    lowerCamelCase_,					[
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			    ],					)
			lowercase__					:      List[Any]      =		audio_classifier(lowerCamelCase_,					top_k=1			)
			self.assertEqual(
			    lowerCamelCase_,					[
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			    ],					)

			self.run_torchaudio(lowerCamelCase_			)

	@require_torchaudio
	def    snake_case__(						self:						Dict,					lowerCamelCase_:						str			):
			import datasets

			# test with a local file
			lowercase__					:      Optional[Any]      =		datasets.load_dataset('hf-internal-testing/librispeech_asr_dummy',					'clean',					split='validation'			)
			lowercase__					:      int      =		dataset[0]['audio']['array']
			lowercase__					:      Tuple      =		audio_classifier(lowerCamelCase_			)
			self.assertEqual(
			    lowerCamelCase_,					[
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			        {'score': ANY(lowerCamelCase_			), 'label': ANY(lowerCamelCase_			)},
			    ],					)

	@require_torch
	def    snake_case__(						self:						Optional[int]			):
			lowercase__					:      List[str]      =		'anton-l/wav2vec2-random-tiny-classifier'

			lowercase__					:      Tuple      =		pipeline('audio-classification',					model=lowerCamelCase_			)

			lowercase__					:      List[str]      =		np.ones((8000,)			)
			lowercase__					:      List[Any]      =		audio_classifier(lowerCamelCase_,					top_k=4			)

			lowercase__					:      List[str]      =		[
			    {'score': 0.0_8_4_2, 'label': 'no'},
			    {'score': 0.0_8_3_8, 'label': 'up'},
			    {'score': 0.0_8_3_7, 'label': 'go'},
			    {'score': 0.0_8_3_4, 'label': 'right'},
			]
			lowercase__					:      Optional[int]      =		[
			    {'score': 0.0_8_4_5, 'label': 'stop'},
			    {'score': 0.0_8_4_4, 'label': 'on'},
			    {'score': 0.0_8_4_1, 'label': 'right'},
			    {'score': 0.0_8_3_4, 'label': 'left'},
			]
			self.assertIn(nested_simplify(lowerCamelCase_,					decimals=4			),					[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]			)

			lowercase__					:      Tuple      =		{'array': np.ones((8000,)			), 'sampling_rate': audio_classifier.feature_extractor.sampling_rate}
			lowercase__					:      Any      =		audio_classifier(lowerCamelCase_,					top_k=4			)
			self.assertIn(nested_simplify(lowerCamelCase_,					decimals=4			),					[EXPECTED_OUTPUT, EXPECTED_OUTPUT_PT_2]			)

	@require_torch
	@slow
	def    snake_case__(						self:						Optional[int]			):
			import datasets

			lowercase__					:      str      =		'superb/wav2vec2-base-superb-ks'

			lowercase__					:      Optional[int]      =		pipeline('audio-classification',					model=lowerCamelCase_			)
			lowercase__					:      List[Any]      =		datasets.load_dataset('anton-l/superb_dummy',					'ks',					split='test'			)

			lowercase__					:      Optional[Any]      =		np.array(dataset[3]['speech'],					dtype=np.floataa			)
			lowercase__					:      Optional[int]      =		audio_classifier(lowerCamelCase_,					top_k=4			)
			self.assertEqual(
			    nested_simplify(lowerCamelCase_,					decimals=3			),					[
			        {'score': 0.9_8_1, 'label': 'go'},
			        {'score': 0.0_0_7, 'label': 'up'},
			        {'score': 0.0_0_6, 'label': '_unknown_'},
			        {'score': 0.0_0_1, 'label': 'down'},
			    ],					)





	@require_tf
	@unittest.skip('Audio classification is not implemented for TF'			)
	def    snake_case__(						self:						int			):
			pass


