


'''simple docstring'''






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       							=				logging.get_logger(__name__)


def    _snake_case  (	)		->       Union[str, Any]:
							# Get the sagemaker specific mp parameters from smp_options variable.
							lowerCAmelCase__			  =      os.getenv('''SM_HP_MP_PARAMETERS'''					,       '''{}'''			)
							try:
														# Parse it and check the field "partitions" is included, it is required for model parallel.
														lowerCAmelCase__			  =      json.loads(A			)
														if "partitions" not in smp_options:
																					return False
							except json.JSONDecodeError:
														return False

							# Get the sagemaker specific framework parameters from mpi_options variable.
							lowerCAmelCase__			  =      os.getenv('''SM_FRAMEWORK_PARAMS'''					,       '''{}'''			)
							try:
														# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
														lowerCAmelCase__			  =      json.loads(A			)
														if not mpi_options.get('''sagemaker_mpi_enabled'''					,       A			):
																					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 						a__							(				a__    ):





							'''simple docstring'''






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







							def     __SCREAMING_SNAKE_CASE				(						self  )			->				Dict:
														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     __SCREAMING_SNAKE_CASE				(						self  )			->				"torch.device":
														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:
																					lowerCAmelCase__			  =      torch.device('''cpu'''  )
																					lowerCAmelCase__			  =      0
														elif is_sagemaker_model_parallel_available():
																					lowerCAmelCase__			  =      smp.local_rank()
																					lowerCAmelCase__			  =      torch.device('''cuda'''   ,       lowerCamelCase_  )
																					lowerCAmelCase__			  =      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  )
																					lowerCAmelCase__			  =      int(os.getenv('''SMDATAPARALLEL_LOCAL_RANK'''  )  )
																					lowerCAmelCase__			  =      torch.device('''cuda'''   ,       self.local_rank  )
																					lowerCAmelCase__			  =      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
																					lowerCAmelCase__			  =      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.
																					lowerCAmelCase__			  =      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  )
																					lowerCAmelCase__			  =      torch.device('''cuda'''   ,       self.local_rank  )
																					lowerCAmelCase__			  =      1

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

														return device







							@property
							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[str]:
														if is_sagemaker_model_parallel_available():
																					return smp.dp_size()

														return super().world_size







							@property
							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[str]:
														return not is_sagemaker_model_parallel_available()







							@property
							def     __SCREAMING_SNAKE_CASE				(						self  )			->				List[Any]:
														return False

import os
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen, xsplitext
from ..table import array_cast
from ..utils.py_utils import no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
							from .features import FeatureType


lowerCamelCase_ ,							lowerCamelCase_ ,							lowerCamelCase_    :			Dict    =	False, False, False

@dataclass
class 							_UpperCamelCase					:






			'''simple docstring'''

			__UpperCamelCase   :    Optional[int]          =			None
			__UpperCamelCase   :    bool          =			True
			__UpperCamelCase   :    bool          =			True
			__UpperCamelCase   :    Optional[str]          =			None
			# Automatically constructed
			__UpperCamelCase   :    ClassVar[str]          =			"dict"
			__UpperCamelCase   :    ClassVar[Any]          =			pa.struct({"""bytes""": pa.binary(), """path""": pa.string()}	)
			__UpperCamelCase   :    str          =			field(default="""Audio"""  ,							init=_A  ,							repr=_A	)

			def __call__(							self			:							List[Any]   ):
							return self.pa_type

			def    lowerCAmelCase__						(							self			:							Optional[int]    ,						snake_case_			:							Union[str, bytes, dict]   ):
							try:
											import soundfile as sf  # soundfile is a dependency of librosa, needed to decode audio files.
							except ImportError as err:
											raise ImportError("""To support encoding audio data, please install 'soundfile'."""   ) from err
							if isinstance(snake_case_    ,						snake_case_   ):
											return {"bytes": None, "path": value}
							elif isinstance(snake_case_    ,						snake_case_   ):
											return {"bytes": value, "path": None}
							elif "array" in value:
											# convert the audio array to wav bytes
											UpperCamelCase_:				List[Any]						=    BytesIO()
											sf.write(snake_case_    ,						value["""array"""]    ,						value["""sampling_rate"""]    ,						format="""wav"""   )
											return {"bytes": buffer.getvalue(), "path": None}
							elif value.get("""path"""   ) is not None and os.path.isfile(value["""path"""]   ):
											# we set "bytes": None to not duplicate the data if they're already available locally
											if value["path"].endswith("""pcm"""   ):
															# "PCM" only has raw audio bytes
															if value.get("""sampling_rate"""   ) is None:
																			# At least, If you want to convert "PCM-byte" to "WAV-byte", you have to know sampling rate
																			raise KeyError("""To use PCM files, please specify a 'sampling_rate' in Audio object"""   )
															if value.get("""bytes"""   ):
																			# If we already had PCM-byte, we don`t have to make "read file, make bytes" (just use it!)
																			UpperCamelCase_:				str						=    np.frombuffer(value["""bytes"""]    ,						dtype=np.intaa   ).astype(np.floataa   ) / 3_2767
															else:
																			UpperCamelCase_:				Dict						=    np.memmap(value["""path"""]    ,						dtype="""h"""    ,						mode="""r"""   ).astype(np.floataa   ) / 3_2767

															UpperCamelCase_:				Dict						=    BytesIO(bytes()   )
															sf.write(snake_case_    ,						snake_case_    ,						value["""sampling_rate"""]    ,						format="""wav"""   )
															return {"bytes": buffer.getvalue(), "path": None}
											else:
															return {"bytes": None, "path": value.get("""path"""   )}
							elif value.get("""bytes"""   ) is not None or value.get("""path"""   ) is not None:
											# store the audio bytes, and path is used to infer the audio format using the file extension
											return {"bytes": value.get("""bytes"""   ), "path": value.get("""path"""   )}
							else:
											raise ValueError(
											    f'''An audio sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'''   )

			def    lowerCAmelCase__						(							self			:							List[str]    ,						snake_case_			:							dict    ,						snake_case_			:							Optional[Dict[str, Union[str, bool, None]]] = None   ):
							if not self.decode:
											raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""   )

							UpperCamelCase_, UpperCamelCase_:				Any						=    (value["""path"""], BytesIO(value["""bytes"""]   )) if value["""bytes"""] is not None else (value["""path"""], None)
							if path is None and file is None:
											raise ValueError(f'''An audio sample should have one of \'path\' or \'bytes\' but both are None in {value}.'''   )

							try:
											import librosa
											import soundfile as sf
							except ImportError as err:
											raise ImportError("""To support decoding audio files, please install 'librosa' and 'soundfile'."""   ) from err

							UpperCamelCase_:				Optional[Any]						=    xsplitext(snake_case_   )[1][1:].lower() if path is not None else None
							if not config.IS_OPUS_SUPPORTED and audio_format == "opus":
											raise RuntimeError(
											    """Decoding 'opus' files requires system library 'libsndfile'>=1.0.31, """
											    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """   )
							elif not config.IS_MP3_SUPPORTED and audio_format == "mp3":
											raise RuntimeError(
											    """Decoding 'mp3' files requires system library 'libsndfile'>=1.1.0, """
											    """You can try to update `soundfile` python library: `pip install \"soundfile>=0.12.1\"`. """   )

							if file is None:
											UpperCamelCase_:				Dict						=    token_per_repo_id or {}
											UpperCamelCase_:				str						=    path.split("""::"""   )[-1]
											try:
															UpperCamelCase_:				Any						=    string_to_dict(snake_case_    ,						config.HUB_DATASETS_URL   )["""repo_id"""]
															UpperCamelCase_:				List[str]						=    token_per_repo_id[repo_id]
											except (ValueError, KeyError):
															UpperCamelCase_:				Any						=    None

											with xopen(snake_case_    ,						"""rb"""    ,						use_auth_token=snake_case_   ) as f:
															UpperCamelCase_, UpperCamelCase_:				List[Any]						=    sf.read(snake_case_   )

							else:
											UpperCamelCase_, UpperCamelCase_:				Union[str, Any]						=    sf.read(snake_case_   )

							UpperCamelCase_:				List[str]						=    array.T
							if self.mono:
											UpperCamelCase_:				int						=    librosa.to_mono(snake_case_   )
							if self.sampling_rate and self.sampling_rate != sampling_rate:
											UpperCamelCase_:				Union[str, Any]						=    librosa.resample(snake_case_    ,						orig_sr=snake_case_    ,						target_sr=self.sampling_rate   )
											UpperCamelCase_:				Dict						=    self.sampling_rate

							return {"path": path, "array": array, "sampling_rate": sampling_rate}

			def    lowerCAmelCase__						(							self			:							Tuple   ):
							from .features import Value

							if self.decode:
											raise ValueError("""Cannot flatten a decoded Audio feature."""   )
							return {
							    "bytes": Value("""binary"""   ),
							    "path": Value("""string"""   ),
							}

			def    lowerCAmelCase__						(							self			:							Any    ,						snake_case_			:							Union[pa.StringArray, pa.StructArray]   ):
							if pa.types.is_string(storage.type   ):
											UpperCamelCase_:				Optional[Any]						=    pa.array([None] * len(snake_case_   )    ,						type=pa.binary()   )
											UpperCamelCase_:				List[str]						=    pa.StructArray.from_arrays([bytes_array, storage]    ,						["""bytes""", """path"""]    ,						mask=storage.is_null()   )
							elif pa.types.is_binary(storage.type   ):
											UpperCamelCase_:				str						=    pa.array([None] * len(snake_case_   )    ,						type=pa.string()   )
											UpperCamelCase_:				List[Any]						=    pa.StructArray.from_arrays([storage, path_array]    ,						["""bytes""", """path"""]    ,						mask=storage.is_null()   )
							elif pa.types.is_struct(storage.type   ) and storage.type.get_all_field_indices("""array"""   ):
											UpperCamelCase_:				Union[str, Any]						=    pa.array([Audio().encode_example(snake_case_   ) if x is not None else None for x in storage.to_pylist()]   )
							elif pa.types.is_struct(storage.type   ):
											if storage.type.get_field_index("""bytes"""   ) >= 0:
															UpperCamelCase_:				Optional[Any]						=    storage.field("""bytes"""   )
											else:
															UpperCamelCase_:				str						=    pa.array([None] * len(snake_case_   )    ,						type=pa.binary()   )
											if storage.type.get_field_index("""path"""   ) >= 0:
															UpperCamelCase_:				List[str]						=    storage.field("""path"""   )
											else:
															UpperCamelCase_:				int						=    pa.array([None] * len(snake_case_   )    ,						type=pa.string()   )
											UpperCamelCase_:				Dict						=    pa.StructArray.from_arrays([bytes_array, path_array]    ,						["""bytes""", """path"""]    ,						mask=storage.is_null()   )
							return array_cast(snake_case_    ,						self.pa_type   )




			def    lowerCAmelCase__						(							self			:							Optional[Any]    ,						snake_case_			:							pa.StructArray   ):

							@no_op_if_value_is_null
							def path_to_bytes(snake_case_			:							Optional[int]   ):
											with xopen(snake_case_    ,						"""rb"""   ) as f:
															UpperCamelCase_:				Optional[int]						=    f.read()
											return bytes_

							UpperCamelCase_:				int						=    pa.array(
							    [
							        (path_to_bytes(x["""path"""]   ) if x["""bytes"""] is None else x["""bytes"""]) if x is not None else None
							        for x in storage.to_pylist()
							    ]    ,						type=pa.binary()    ,						)
							UpperCamelCase_:				Union[str, Any]						=    pa.array(
							    [os.path.basename(snake_case_   ) if path is not None else None for path in storage.field("""path"""   ).to_pylist()]    ,						type=pa.string()    ,						)
							UpperCamelCase_:				Optional[int]						=    pa.StructArray.from_arrays([bytes_array, path_array]    ,						["""bytes""", """path"""]    ,						mask=bytes_array.is_null()   )
							return array_cast(snake_case_    ,						self.pa_type   )






