





'''simple docstring'''






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


A__   ,						A__   ,						A__				:   Union[str, Any]      =False, False, False




@dataclass
class  UpperCAmelCase     :
      _lowercase:			Optional[int]						     =   None
      _lowercase:			bool						     =   True
      _lowercase:			bool						     =   True
      _lowercase:			Optional[str]						     =   None
      # Automatically constructed
      _lowercase:			ClassVar[str]						     =   "dict"
      _lowercase:			ClassVar[Any]						     =   pa.struct({'''bytes''': pa.binary(), '''path''': pa.string()}   )
      _lowercase:			str						     =   field(default='''Audio'''							,		init=snake_case_							,		repr=snake_case_   )




      def __call__(			self			: int  )				->      int:
            return self.pa_type




      def 				lowercase__    (			self			: List[Any]			,			__snake_case			: Union[str, bytes, dict]  )				->      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
                  _lowerCAmelCase							     =							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!)
                              _lowerCAmelCase							     =							np.frombuffer(value["""bytes"""]			,			dtype=np.intaa  ).astype(np.floataa  ) / 3_27_67
                        else:
                              _lowerCAmelCase							     =							np.memmap(value["""path"""]			,			dtype="""h"""			,			mode="""r"""  ).astype(np.floataa  ) / 3_27_67

                        _lowerCAmelCase							     =							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 				lowercase__    (			self			: List[Any]			,			__snake_case			: dict			,			__snake_case			: Optional[Dict[str, Union[str, bool, None]]] = None  )				->      dict:
            if not self.decode:
                  raise RuntimeError("""Decoding is disabled for this feature. Please use Audio(decode=True) instead."""  )

            _lowerCAmelCase     ,							_lowerCAmelCase							     =							(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

            _lowerCAmelCase							     =							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:
                  _lowerCAmelCase							     =							token_per_repo_id or {}
                  _lowerCAmelCase							     =							path.split("""::"""  )[-1]
                  try:
                        _lowerCAmelCase							     =							string_to_dict(__snake_case			,			config.HUB_DATASETS_URL  )["""repo_id"""]
                        _lowerCAmelCase							     =							token_per_repo_id[repo_id]
                  except (ValueError, KeyError):
                        _lowerCAmelCase							     =							None

                  with xopen(__snake_case			,			"""rb"""			,			use_auth_token=__snake_case  ) as f:
                        _lowerCAmelCase     ,							_lowerCAmelCase							     =							sf.read(__snake_case  )

            else:
                  _lowerCAmelCase     ,							_lowerCAmelCase							     =							sf.read(__snake_case  )

            _lowerCAmelCase							     =							array.T
            if self.mono:
                  _lowerCAmelCase							     =							librosa.to_mono(__snake_case  )
            if self.sampling_rate and self.sampling_rate != sampling_rate:
                  _lowerCAmelCase							     =							librosa.resample(__snake_case			,			orig_sr=__snake_case			,			target_sr=self.sampling_rate  )
                  _lowerCAmelCase							     =							self.sampling_rate

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




      def 				lowercase__    (			self			: List[str]  )				->      Union["FeatureType", Dict[str, "FeatureType"]]:
            from .features import Value

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




      def 				lowercase__    (			self			: int			,			__snake_case			: Union[pa.StringArray, pa.StructArray]  )				->      pa.StructArray:
            if pa.types.is_string(storage.type  ):
                  _lowerCAmelCase							     =							pa.array([None] * len(__snake_case  )			,			type=pa.binary()  )
                  _lowerCAmelCase							     =							pa.StructArray.from_arrays([bytes_array, storage]			,			["""bytes""", """path"""]			,			mask=storage.is_null()  )
            elif pa.types.is_binary(storage.type  ):
                  _lowerCAmelCase							     =							pa.array([None] * len(__snake_case  )			,			type=pa.string()  )
                  _lowerCAmelCase							     =							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"""  ):
                  _lowerCAmelCase							     =							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:
                        _lowerCAmelCase							     =							storage.field("""bytes"""  )
                  else:
                        _lowerCAmelCase							     =							pa.array([None] * len(__snake_case  )			,			type=pa.binary()  )
                  if storage.type.get_field_index("""path"""  ) >= 0:
                        _lowerCAmelCase							     =							storage.field("""path"""  )
                  else:
                        _lowerCAmelCase							     =							pa.array([None] * len(__snake_case  )			,			type=pa.string()  )
                  _lowerCAmelCase							     =							pa.StructArray.from_arrays([bytes_array, path_array]			,			["""bytes""", """path"""]			,			mask=storage.is_null()  )
            return array_cast(__snake_case			,			self.pa_type  )




      def 				lowercase__    (			self			: Any			,			__snake_case			: pa.StructArray  )				->      pa.StructArray:

            @no_op_if_value_is_null
            def path_to_bytes(__snake_case			: List[Any]  ):
                  with xopen(__snake_case			,			"""rb"""  ) as f:
                        _lowerCAmelCase							     =							f.read()
                  return bytes_

            _lowerCAmelCase							     =							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()			,			)
            _lowerCAmelCase							     =							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()			,			)
            _lowerCAmelCase							     =							pa.StructArray.from_arrays([bytes_array, path_array]			,			["""bytes""", """path"""]			,			mask=bytes_array.is_null()  )
            return array_cast(__snake_case			,			self.pa_type  )








# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from pathlib import Path

import torch

from ...utils import is_npu_available, is_xpu_available
from .config_args import ClusterConfig, default_json_config_file
from .config_utils import SubcommandHelpFormatter


UpperCAmelCase	      = '''Create a default config file for Accelerate with only a few flags set.'''






def 							UpperCAmelCase_		(							__SCREAMING_SNAKE_CASE="no" ,							__SCREAMING_SNAKE_CASE = default_json_config_file ,							__SCREAMING_SNAKE_CASE = False							):
	lowercase       =    Path(__SCREAMING_SNAKE_CASE							)
	path.parent.mkdir(parents=__SCREAMING_SNAKE_CASE ,							exist_ok=__SCREAMING_SNAKE_CASE							)
	if path.exists():
		print(
		    F'''Configuration already exists at {save_location}, will not override. Run `accelerate config` manually or pass a different `save_location`.'''							)
		return False
	lowercase       =    mixed_precision.lower()
	if mixed_precision not in ["no", "fp16", "bf16", "fp8"]:
		raise ValueError(
		    F'''`mixed_precision` should be one of \'no\', \'fp16\', \'bf16\', or \'fp8\'. Received {mixed_precision}'''							)
	lowercase       =    {
	    'compute_environment': 'LOCAL_MACHINE',
	    'mixed_precision': mixed_precision,
	}
	if torch.cuda.is_available():
		lowercase       =    torch.cuda.device_count()
		lowercase       =    num_gpus
		lowercase       =    False
		if num_gpus > 1:
			lowercase       =    'MULTI_GPU'
		else:
			lowercase       =    'NO'
	elif is_xpu_available() and use_xpu:
		lowercase       =    torch.xpu.device_count()
		lowercase       =    num_xpus
		lowercase       =    False
		if num_xpus > 1:
			lowercase       =    'MULTI_XPU'
		else:
			lowercase       =    'NO'
	elif is_npu_available():
		lowercase       =    torch.npu.device_count()
		lowercase       =    num_npus
		lowercase       =    False
		if num_npus > 1:
			lowercase       =    'MULTI_NPU'
		else:
			lowercase       =    'NO'
	else:
		lowercase       =    0
		lowercase       =    True
		lowercase       =    1
		lowercase       =    'NO'
	lowercase       =    ClusterConfig(**__SCREAMING_SNAKE_CASE							)
	config.to_json_file(__SCREAMING_SNAKE_CASE							)
	return path






def 							UpperCAmelCase_		(							__SCREAMING_SNAKE_CASE ,							__SCREAMING_SNAKE_CASE							):
	lowercase       =    parser.add_parser('default' ,							parents=__SCREAMING_SNAKE_CASE ,							help=__SCREAMING_SNAKE_CASE ,							formatter_class=__SCREAMING_SNAKE_CASE							)
	parser.add_argument(
	    '--config_file' ,							default=__SCREAMING_SNAKE_CASE ,							help=(
	        'The path to use to store the config file. Will default to a file named default_config.yaml in the cache '
	        'location, which is the content of the environment `HF_HOME` suffixed with \'accelerate\', or if you don\'t have '
	        'such an environment variable, your cache directory (\'~/.cache\' or the content of `XDG_CACHE_HOME`) suffixed '
	        'with \'huggingface\'.'
	    ) ,							dest='save_location' ,							)

	parser.add_argument(
	    '--mixed_precision' ,							choices=['no', 'fp16', 'bf16'] ,							type=__SCREAMING_SNAKE_CASE ,							help='Whether or not to use mixed precision training. '
	    'Choose between FP16 and BF16 (bfloat16) training. '
	    'BF16 training is only supported on Nvidia Ampere GPUs and PyTorch 1.10 or later.' ,							default='no' ,							)
	parser.set_defaults(func=__SCREAMING_SNAKE_CASE							)
	return parser






def 							UpperCAmelCase_		(							__SCREAMING_SNAKE_CASE							):
	lowercase       =    write_basic_config(args.mixed_precision ,							args.save_location							)
	if config_file:
		print(F'''accelerate configuration saved at {config_file}'''							)

