





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


UpperCAmelCase_					:   Tuple 							=						logging.get_logger(__name__)





class      lowercase__    (							__A  ):
  __UpperCamelCase	     =  ["""input_features"""]







  def __init__(	self    ,						_lowercase=80    ,						_lowercase=16_000    ,						_lowercase=160    ,						_lowercase=30    ,						_lowercase=400    ,						_lowercase=0.0    ,						_lowercase=False    ,						**_lowercase    ,						):
       super().__init__(
           feature_size=_lowercase    ,						sampling_rate=_lowercase    ,						padding_value=_lowercase    ,						return_attention_mask=_lowercase    ,						**_lowercase    ,						)
       lowerCAmelCase_					:       str    =		n_fft
       lowerCAmelCase_					:       Any    =		hop_length
       lowerCAmelCase_					:       Union[str, Any]    =		chunk_length
       lowerCAmelCase_					:       Dict    =		chunk_length * sampling_rate
       lowerCAmelCase_					:       Tuple    =		self.n_samples // hop_length
       lowerCAmelCase_					:       Optional[int]    =		sampling_rate
       lowerCAmelCase_					:       str    =		mel_filter_bank(
           num_frequency_bins=1 + n_fft // 2    ,						num_mel_filters=_lowercase    ,						min_frequency=0.0    ,						max_frequency=8000.0    ,						sampling_rate=_lowercase    ,						norm="""slaney"""    ,						mel_scale="""slaney"""    ,						)







  def 	UpperCAmelCase__		(	self    ,						_lowercase						):
       lowerCAmelCase_					:       Any    =		spectrogram(
           _lowercase    ,						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"""    ,						)
       lowerCAmelCase_					:       Optional[Any]    =		log_spec[:, :-1]
       lowerCAmelCase_					:       Dict    =		np.maximum(_lowercase    ,						log_spec.max() - 8.0						)
       lowerCAmelCase_					:       Optional[int]    =		(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 	UpperCAmelCase__		(	_lowercase    ,						_lowercase    ,						_lowercase = 0.0						):
       if attention_mask is not None:
            lowerCAmelCase_					:       str    =		np.array(_lowercase    ,						np.intaa						)
            lowerCAmelCase_					:       str    =		[]

            for vector, length in zip(_lowercase    ,						attention_mask.sum(-1						)						):
                 lowerCAmelCase_					:       Tuple    =		(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1e-7						)
                 if length < normed_slice.shape[0]:
                      lowerCAmelCase_					:       Dict    =		padding_value

                 normed_input_values.append(_lowercase						)
       else:
            lowerCAmelCase_					:       List[str]    =		[(x - x.mean()) / np.sqrt(x.var() + 1e-7						) for x in input_values]

       return normed_input_values







  def __call__(	self    ,						_lowercase    ,						_lowercase = True    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase = "max_length"    ,						_lowercase = None    ,						_lowercase = None    ,						_lowercase = None    ,						**_lowercase    ,						):

       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."""						)

       lowerCAmelCase_					:       Any    =		isinstance(_lowercase    ,						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}'						)
       lowerCAmelCase_					:       Union[str, Any]    =		is_batched_numpy or (
           isinstance(_lowercase    ,						(list, tuple)						) and (isinstance(raw_speech[0]    ,						(np.ndarray, tuple, list)						))
       )

       if is_batched:
            lowerCAmelCase_					:       List[Any]    =		[np.asarray([speech]    ,						dtype=np.floataa						).T for speech in raw_speech]
       elif not is_batched and not isinstance(_lowercase    ,						np.ndarray						):
            lowerCAmelCase_					:       str    =		np.asarray(_lowercase    ,						dtype=np.floataa						)
       elif isinstance(_lowercase    ,						np.ndarray						) and raw_speech.dtype is np.dtype(np.floataa						):
            lowerCAmelCase_					:       Optional[Any]    =		raw_speech.astype(np.floataa						)

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

       lowerCAmelCase_					:       List[str]    =		BatchFeature({"""input_features""": raw_speech}						)

       # convert into correct format for padding

       lowerCAmelCase_					:       str    =		self.pad(
           _lowercase    ,						padding=_lowercase    ,						max_length=max_length if max_length else self.n_samples    ,						truncation=_lowercase    ,						pad_to_multiple_of=_lowercase    ,						return_attention_mask=return_attention_mask or do_normalize    ,						)

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

       # make sure list is in array format
       lowerCAmelCase_					:       List[str]    =		padded_inputs.get("""input_features"""						).transpose(2    ,						0    ,						1						)

       lowerCAmelCase_					:       Optional[int]    =		[self._np_extract_fbank_features(_lowercase						) for waveform in input_features[0]]

       if isinstance(input_features[0]    ,						_lowercase						):
            lowerCAmelCase_					:       Tuple    =		[np.asarray(_lowercase    ,						dtype=np.floataa						) for feature in input_features]
       else:
            lowerCAmelCase_					:       str    =		input_features

       if return_attention_mask:
            # rescale from sample (48000) to feature (3000)
            lowerCAmelCase_					:       Union[str, Any]    =		padded_inputs["""attention_mask"""][:, :: self.hop_length]

       if return_tensors is not None:
            lowerCAmelCase_					:       Dict    =		padded_inputs.convert_to_tensors(_lowercase						)

       return padded_inputs







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       int    =		copy.deepcopy(self.__dict__						)
       lowerCAmelCase_					:       Union[str, Any]    =		self.__class__.__name__
       if "mel_filters" in output:
            del output["mel_filters"]
       return output







import inspect
import os
import unittest
from pathlib import Path

import torch

import accelerate
from accelerate.test_utils import execute_subprocess_async
from accelerate.test_utils.testing import run_command





class      lowercase__    (							unittest.TestCase  ):
  __UpperCamelCase	     =  inspect.getfile(accelerate.test_utils  )
  __UpperCamelCase	     =  os.path.sep.join(mod_file.split(os.path.sep  )[:-1] + ["""scripts""", """test_cli.py"""]  )

  __UpperCamelCase	     =  ["""accelerate""", """launch"""]
  __UpperCamelCase	     =  Path.home() / """.cache/huggingface/accelerate"""
  __UpperCamelCase	     =  """default_config.yaml"""
  __UpperCamelCase	     =  config_folder / config_file
  __UpperCamelCase	     =  config_folder / """_default_config.yaml"""

  __UpperCamelCase	     =  Path("""tests/test_configs"""  )







  @classmethod
  def 	UpperCAmelCase__		(	cls						):
       if cls.config_path.is_file():
            cls.config_path.rename(cls.changed_path						)







  @classmethod
  def 	UpperCAmelCase__		(	cls						):
       if cls.changed_path.is_file():
            cls.changed_path.rename(cls.config_path						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       int    =		self.base_cmd
       if torch.cuda.is_available() and (torch.cuda.device_count() > 1):
            cmd += ["--multi_gpu"]
       execute_subprocess_async(cmd + [self.test_file_path]    ,						env=os.environ.copy()						)







  def 	UpperCAmelCase__		(	self						):
       for config in sorted(self.test_config_path.glob("""**/*.yaml"""						)						):
            with self.subTest(config_file=_lowercase						):
                 execute_subprocess_async(
                     self.base_cmd + ["""--config_file""", str(_lowercase						), self.test_file_path]    ,						env=os.environ.copy()						)







  def 	UpperCAmelCase__		(	self						):
       execute_subprocess_async(["""accelerate""", """test"""]    ,						env=os.environ.copy()						)





class      lowercase__    (							unittest.TestCase  ):
  __UpperCamelCase	     =  """test-tpu"""
  __UpperCamelCase	     =  """us-central1-a"""
  __UpperCamelCase	     =  """ls"""
  __UpperCamelCase	     =  ["""accelerate""", """tpu-config"""]
  __UpperCamelCase	     =  """cd /usr/share"""
  __UpperCamelCase	     =  """tests/test_samples/test_command_file.sh"""
  __UpperCamelCase	     =  """Running gcloud compute tpus tpu-vm ssh"""







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       str    =		run_command(
           self.cmd
           + ["""--command""", self.command, """--tpu_zone""", self.tpu_zone, """--tpu_name""", self.tpu_name, """--debug"""]    ,						return_stdout=_lowercase    ,						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'    ,						_lowercase    ,						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       str    =		run_command(
           self.cmd
           + [
               """--config_file""",
               """tests/test_configs/0_12_0.yaml""",
               """--command""",
               self.command,
               """--tpu_zone""",
               self.tpu_zone,
               """--tpu_name""",
               self.tpu_name,
               """--debug""",
           ]    ,						return_stdout=_lowercase    ,						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'    ,						_lowercase    ,						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       Optional[Any]    =		run_command(
           self.cmd + ["""--config_file""", """tests/test_configs/latest.yaml""", """--debug"""]    ,						return_stdout=_lowercase						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'    ,						_lowercase    ,						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       Tuple    =		run_command(
           self.cmd + ["""--config_file""", """tests/test_configs/latest.yaml""", """--command""", self.command, """--debug"""]    ,						return_stdout=_lowercase    ,						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls --worker all'    ,						_lowercase    ,						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       str    =		run_command(
           self.cmd
           + [
               """--config_file""",
               """tests/test_configs/latest.yaml""",
               """--command""",
               self.command,
               """--command""",
               """echo \"Hello World\"""",
               """--debug""",
           ]    ,						return_stdout=_lowercase    ,						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; ls; echo "Hello World" --worker all'    ,						_lowercase    ,						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       str    =		run_command(
           self.cmd
           + ["""--config_file""", """tests/test_configs/latest.yaml""", """--command_file""", self.command_file, """--debug"""]    ,						return_stdout=_lowercase    ,						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'    ,						_lowercase    ,						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       Optional[Any]    =		run_command(
           self.cmd
           + [
               """--config_file""",
               """tests/test_configs/0_12_0.yaml""",
               """--command_file""",
               self.command_file,
               """--tpu_zone""",
               self.tpu_zone,
               """--tpu_name""",
               self.tpu_name,
               """--debug""",
           ]    ,						return_stdout=_lowercase    ,						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; echo "hello world"; echo "this is a second command" --worker all'    ,						_lowercase    ,						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       Tuple    =		run_command(
           self.cmd + ["""--config_file""", """tests/test_configs/latest.yaml""", """--install_accelerate""", """--debug"""]    ,						return_stdout=_lowercase    ,						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate -U; echo "hello world"; echo "this is a second command" --worker all'    ,						_lowercase    ,						)







  def 	UpperCAmelCase__		(	self						):
       lowerCAmelCase_					:       Any    =		run_command(
           self.cmd
           + [
               """--config_file""",
               """tests/test_configs/latest.yaml""",
               """--install_accelerate""",
               """--accelerate_version""",
               """12.0.0""",
               """--debug""",
           ]    ,						return_stdout=_lowercase    ,						)
       self.assertIn(
           F'{self.gcloud} test-tpu --zone us-central1-a --command {self.base_output}; pip install accelerate==12.0.0; echo "hello world"; echo "this is a second command" --worker all'    ,						_lowercase    ,						)
