"""simple docstring"""





import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import TensorType, logging


_a							=							logging.get_logger(__name__)




class  _lowerCAmelCase  (     lowerCamelCase__       ):







 """simple docstring"""







 __UpperCAmelCase     : str       		=     ["input_features"]


 def __init__(      self							:      Any, UpperCAmelCase__							:      Optional[int]=8_0, UpperCAmelCase__							:      List[str]=1_6_0_0_0, UpperCAmelCase__							:      Any=1_6_0, UpperCAmelCase__							:      Union[str, Any]=3_0, UpperCAmelCase__							:      Optional[Any]=4_0_0, UpperCAmelCase__							:      Optional[int]=0.0, UpperCAmelCase__							:      Optional[Any]=False, **UpperCAmelCase__							:      List[Any], ):
    super().__init__(
        feature_size=__A, sampling_rate=__A, padding_value=__A, return_attention_mask=__A, **__A, )
    __lowercase								=						n_fft
    __lowercase								=						hop_length
    __lowercase								=						chunk_length
    __lowercase								=						chunk_length * sampling_rate
    __lowercase								=						self.n_samples // hop_length
    __lowercase								=						sampling_rate
    __lowercase								=						mel_filter_bank(
        num_frequency_bins=1 + n_fft // 2, num_mel_filters=__A, min_frequency=0.0, max_frequency=8_0_0_0.0, sampling_rate=__A, norm="slaney", mel_scale="slaney", )


 def  _lowercase			(      self							:      Tuple, UpperCAmelCase__							:      np.array     ):
    __lowercase								=						spectrogram(
        __A, window_function(self.n_fft, "hann"     ), frame_length=self.n_fft, hop_length=self.hop_length, power=2.0, mel_filters=self.mel_filters, log_mel="log10", )
    __lowercase								=						log_spec[:, :-1]
    __lowercase								=						np.maximum(__A, log_spec.max() - 8.0     )
    __lowercase								=						(log_spec + 4.0) / 4.0
    return log_spec


 @staticmethod
 # Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
 def  _lowercase			(      UpperCAmelCase__							:      List[np.ndarray], UpperCAmelCase__							:      List[np.ndarray], UpperCAmelCase__							:      float = 0.0     ):
    if attention_mask is not None:
       __lowercase								=						np.array(__A, np.intaa     )
       __lowercase								=						[]

       for vector, length in zip(__A, attention_mask.sum(-1     )     ):
          __lowercase								=						(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7     )
          if length < normed_slice.shape[0]:
             __lowercase								=						padding_value

          normed_input_values.append(__A     )
    else:
       __lowercase								=						[(x - x.mean()) / np.sqrt(x.var() + 1E-7     ) for x in input_values]

    return normed_input_values


 def __call__(      self							:      Any, UpperCAmelCase__							:      Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]], UpperCAmelCase__							:      bool = True, UpperCAmelCase__							:      Optional[int] = None, UpperCAmelCase__							:      Optional[Union[str, TensorType]] = None, UpperCAmelCase__							:      Optional[bool] = None, UpperCAmelCase__							:      Optional[str] = "max_length", UpperCAmelCase__							:      Optional[int] = None, UpperCAmelCase__							:      Optional[int] = None, UpperCAmelCase__							:      Optional[bool] = None, **UpperCAmelCase__							:      Tuple, ):

    if sampling_rate is not None:
       if sampling_rate != self.sampling_rate:
          raise ValueError(
              F"""The model corresponding to this feature extractor: {self.__class__.__name__} was trained using a"""
              F""" sampling rate of {self.sampling_rate}. Please make sure that the provided `raw_speech` input"""
              F""" was sampled with {self.sampling_rate} and not {sampling_rate}."""     )
    else:
       logger.warning(
           "It is strongly recommended to pass the `sampling_rate` argument to this function. "
           "Failing to do so can result in silent errors that might be hard to debug."     )

    __lowercase								=						isinstance(__A, np.ndarray     ) and len(raw_speech.shape     ) > 1
    if is_batched_numpy and len(raw_speech.shape     ) > 2:
       raise ValueError(F"""Only mono-channel audio is supported for input to {self}"""     )
    __lowercase								=						is_batched_numpy or (
        isinstance(__A, (list, tuple)     ) and (isinstance(raw_speech[0], (np.ndarray, tuple, list)     ))
    )

    if is_batched:
       __lowercase								=						[np.asarray([speech], dtype=np.floataa     ).T for speech in raw_speech]
    elif not is_batched and not isinstance(__A, np.ndarray     ):
       __lowercase								=						np.asarray(__A, dtype=np.floataa     )
    elif isinstance(__A, np.ndarray     ) and raw_speech.dtype is np.dtype(np.floataa     ):
       __lowercase								=						raw_speech.astype(np.floataa     )

    # always return batch
    if not is_batched:
       __lowercase								=						[np.asarray([raw_speech]     ).T]

    __lowercase								=						BatchFeature({"input_features": raw_speech}     )

    # convert into correct format for padding

    __lowercase								=						self.pad(
        __A, padding=__A, max_length=max_length if max_length else self.n_samples, truncation=__A, pad_to_multiple_of=__A, return_attention_mask=return_attention_mask or do_normalize, )

    # zero-mean and unit-variance normalization
    if do_normalize:
       __lowercase								=						self.zero_mean_unit_var_norm(
           padded_inputs["input_features"], attention_mask=padded_inputs["attention_mask"], padding_value=self.padding_value, )
       __lowercase								=						np.stack(padded_inputs["input_features"], axis=0     )

    # make sure list is in array format
    __lowercase								=						padded_inputs.get("input_features"     ).transpose(2, 0, 1     )

    __lowercase								=						[self._np_extract_fbank_features(__A     ) for waveform in input_features[0]]

    if isinstance(input_features[0], __A     ):
       __lowercase								=						[np.asarray(__A, dtype=np.floataa     ) for feature in input_features]
    else:
       __lowercase								=						input_features

    if return_attention_mask:
       # rescale from sample (48000) to feature (3000)
       __lowercase								=						padded_inputs['''attention_mask'''][:, :: self.hop_length]

    if return_tensors is not None:
       __lowercase								=						padded_inputs.convert_to_tensors(__A     )

    return padded_inputs



 def  _lowercase			(      self							:      List[Any]     ):
    __lowercase								=						copy.deepcopy(self.__dict__     )
    __lowercase								=						self.__class__.__name__
    if "mel_filters" in output:
       del output["mel_filters"]
    return output







"""simple docstring"""





# 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.

import argparse
import os

from accelerate.utils import ComputeEnvironment

from .cluster import get_cluster_input
from .config_args import cache_dir, default_config_file, default_yaml_config_file, load_config_from_file  # noqa: F401
from .config_utils import _ask_field, _ask_options, _convert_compute_environment  # noqa: F401
from .sagemaker import get_sagemaker_input


_a							=							'Launches a series of prompts to create and save a `default_config.yaml` configuration file for your training system. Should always be ran first on your machine'




def   _A  (  )     ->						Tuple:
   '''simple docstring'''


   __lowercase								=						_ask_options(
       "In which compute environment are you running?",      ["This machine", "AWS (Amazon SageMaker)"],      _convert_compute_environment,      )
   if compute_environment == ComputeEnvironment.AMAZON_SAGEMAKER:
      __lowercase								=						get_sagemaker_input()
   else:
      __lowercase								=						get_cluster_input()
   return config




def   _A  (  UpperCamelCase_					:	Union[str, Any]=None)     ->						Union[str, Any]:
   '''simple docstring'''


   if subparsers is not None:
      __lowercase								=						subparsers.add_parser("config",      description=UpperCamelCase_)
   else:
      __lowercase								=						argparse.ArgumentParser("Accelerate config command",      description=UpperCamelCase_)

   parser.add_argument(
       "--config_file",      default=UpperCamelCase_,      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'."
       ),      )

   if subparsers is not None:
      parser.set_defaults(func=UpperCamelCase_)
   return parser




def   _A  (  UpperCamelCase_					:	Dict)     ->						str:
   '''simple docstring'''


   __lowercase								=						get_user_input()
   if args.config_file is not None:
      __lowercase								=						args.config_file
   else:
      if not os.path.isdir(UpperCamelCase_):
         os.makedirs(UpperCamelCase_)
      __lowercase								=						default_yaml_config_file

   if config_file.endswith(".json"):
      config.to_json_file(UpperCamelCase_)
   else:
      config.to_yaml_file(UpperCamelCase_)
   print(F"""accelerate configuration saved at {config_file}""")




def   _A  (  )     ->						Optional[Any]:
   '''simple docstring'''


   __lowercase								=						config_command_parser()
   __lowercase								=						parser.parse_args()
   config_command(UpperCamelCase_)


if __name__ == "__main__":
   main()




