


'''simple docstring'''






# Copyright 2022 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
import subprocess

from packaging.version import Version, parse

from accelerate.commands.config.config_args import default_config_file, load_config_from_file


a_   :      Dict				    =      "Run commands across TPU VMs for initial setup before running `accelerate launch`."






def 		_A			(lowerCAmelCase__					:Any=None						)				->					List[Any]:




     '''simple docstring'''



     if subparsers is not None:
          _a      =							subparsers.add_parser('tpu-config'			,  description=_description						)
     else:
          _a      =							argparse.ArgumentParser('Accelerate tpu-config command'			,  description=_description						)
     # Core arguments
     _a      =							parser.add_argument_group(
         'Config Arguments'			,  'Arguments that can be configured through `accelerate config`.'						)
     config_args.add_argument(
         '--config_file'			,  type=lowerCAmelCase__			,  default=lowerCAmelCase__			,  help='Path to the config file to use for accelerate.'			,  )
     config_args.add_argument(
         '--tpu_name'			,  default=lowerCAmelCase__			,  help='The name of the TPU to use. If not specified, will use the TPU specified in the config file.'			,  )
     config_args.add_argument(
         '--tpu_zone'			,  default=lowerCAmelCase__			,  help='The zone of the TPU to use. If not specified, will use the zone specified in the config file.'			,  )
     _a      =							parser.add_argument_group('TPU Arguments'			,  'Arguments for options ran inside the TPU.'						)
     pod_args.add_argument(
         '--use_alpha'			,  action='store_true'			,  help='Whether to use `gcloud alpha` when running the TPU training script instead of `gcloud`.'			,  )
     pod_args.add_argument(
         '--command_file'			,  default=lowerCAmelCase__			,  help='The path to the file containing the commands to run on the pod on startup.'			,  )
     pod_args.add_argument(
         '--command'			,  action='append'			,  nargs='+'			,  help='A command to run on the pod. Can be passed multiple times.'			,  )
     pod_args.add_argument(
         '--install_accelerate'			,  action='store_true'			,  help='Whether to install accelerate on the pod. Defaults to False.'			,  )
     pod_args.add_argument(
         '--accelerate_version'			,  default='latest'			,  help='The version of accelerate to install on the pod. If not specified, will use the latest pypi version. Specify \'dev\' to install from GitHub.'			,  )
     pod_args.add_argument(
         '--debug'			,  action='store_true'			,  help='If set, will print the command that would be run instead of running it.'						)

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






def 		_A			(lowerCAmelCase__					:str						)				->					str:




     '''simple docstring'''



     _a      =							None

     # Get the default from the config file if it exists.
     if args.config_file is not None or os.path.isfile(lowerCAmelCase__						):
          _a      =							load_config_from_file(args.config_file						)
          if not args.command_file and defaults.command_file is not None and not args.command:
               _a      =							defaults.command_file
          if not args.command and defaults.commands is not None:
               _a      =							defaults.commands
          if not args.tpu_name:
               _a      =							defaults.tpu_name
          if not args.tpu_zone:
               _a      =							defaults.tpu_zone
     if args.accelerate_version == "dev":
          _a      =							'git+https://github.com/huggingface/accelerate.git'
     elif args.accelerate_version == "latest":
          _a      =							'accelerate -U'
     elif isinstance(parse(args.accelerate_version						)			,  lowerCAmelCase__						):
          _a      =							f'accelerate=={args.accelerate_version}'

     if not args.command_file and not args.command:
          raise ValueError('You must specify either a command file or a command to run on the pod.'						)

     if args.command_file:
          with open(args.command_file			,  'r'						) as f:
               _a      =							[f.read().splitlines()]

    # To turn list of lists into list of strings
     if isinstance(args.command[0]			,  lowerCAmelCase__						):
          _a      =							[line for cmd in args.command for line in cmd]
     # Default to the shared folder and install accelerate
     _a      =							['cd /usr/share']
     if args.install_accelerate:
          new_cmd += [f'pip install {args.accelerate_version}']
     new_cmd += args.command
     _a      =							'; '.join(lowerCAmelCase__						)

     # Then send it to gcloud
     # Eventually try to use google-api-core to do this instead of subprocess
     _a      =							['gcloud']
     if args.use_alpha:
          cmd += ["alpha"]
     cmd += [
         "compute",
         "tpus",
         "tpu-vm",
         "ssh",
         args.tpu_name,
         "--zone",
         args.tpu_zone,
         "--command",
         args.command,
         "--worker",
         "all",
     ]
     if args.debug:
          print(f'Running {" ".join(lowerCAmelCase__						)}'						)
          return
     subprocess.run(lowerCAmelCase__						)
     print('Successfully setup pod.'						)






def 		_A			()				->					List[str]:




     '''simple docstring'''



     _a      =							tpu_command_parser()
     _a      =							parser.parse_args()

     tpu_command_launcher(lowerCAmelCase__						)





'''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_   :      Tuple				    =      logging.get_logger(__name__)




class 	a       (       _SCREAMING_SNAKE_CASE    ):
 _lowerCAmelCase	   =		["""input_features"""]
 def __init__(	self ,			__magic_name__=80 ,			__magic_name__=1_60_00 ,			__magic_name__=1_60 ,			__magic_name__=30 ,			__magic_name__=4_00 ,			__magic_name__=0.0 ,			__magic_name__=False ,			**__magic_name__ ,			)   ->							Optional[int]:
      super().__init__(
          feature_size=__magic_name__ ,			sampling_rate=__magic_name__ ,			padding_value=__magic_name__ ,			return_attention_mask=__magic_name__ ,			**__magic_name__ ,			)
      _a      =							n_fft
      _a      =							hop_length
      _a      =							chunk_length
      _a      =							chunk_length * sampling_rate
      _a      =							self.n_samples // hop_length
      _a      =							sampling_rate
      _a      =							mel_filter_bank(
          num_frequency_bins=1 + n_fft // 2 ,			num_mel_filters=__magic_name__ ,			min_frequency=0.0 ,			max_frequency=8_0_0_0.0 ,			sampling_rate=__magic_name__ ,			norm='slaney' ,			mel_scale='slaney' ,			)
 def 							__UpperCAmelCase						(	self ,			__magic_name__  )   ->							np.ndarray:
      _a      =							spectrogram(
          __magic_name__ ,			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' ,			)
      _a      =							log_spec[:, :-1]
      _a      =							np.maximum(__magic_name__ ,			log_spec.max() - 8.0  )
      _a      =							(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						(	__magic_name__ ,			__magic_name__ ,			__magic_name__ = 0.0  )   ->							List[np.ndarray]:
      if attention_mask is not None:
           _a      =							np.array(__magic_name__ ,			np.intaa  )
           _a      =							[]

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

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

      return normed_input_values
 def __call__(	self ,			__magic_name__ ,			__magic_name__ = True ,			__magic_name__ = None ,			__magic_name__ = None ,			__magic_name__ = None ,			__magic_name__ = "max_length" ,			__magic_name__ = None ,			__magic_name__ = None ,			__magic_name__ = None ,			**__magic_name__ ,			)   ->							BatchFeature:

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

      _a      =							isinstance(__magic_name__ ,			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}'  )
      _a      =							is_batched_numpy or (
          isinstance(__magic_name__ ,			(list, tuple)  ) and (isinstance(raw_speech[0] ,			(np.ndarray, tuple, list)  ))
      )

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

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

      _a      =							BatchFeature({'input_features': raw_speech}  )

      # convert into correct format for padding

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

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

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

      _a      =							[self._np_extract_fbank_features(__magic_name__  ) for waveform in input_features[0]]

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

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

      if return_tensors is not None:
           _a      =							padded_inputs.convert_to_tensors(__magic_name__  )

      return padded_inputs




 def 							__UpperCAmelCase						(	self  )   ->							Dict[str, Any]:
      _a      =							copy.deepcopy(self.__dict__  )
      _a      =							self.__class__.__name__
      if "mel_filters" in output:
           del output["mel_filters"]
      return output

