





import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map



def 						A_							(	_lowerCAmelCase	)		->					List[Tuple[int, ...]]:
 UpperCamelCase      :   Optional[Any]			      =  []
 if isinstance(_lowerCAmelCase				,  _lowerCAmelCase	):
  for v in tree.values():
   shapes.extend(_fetch_dims(_lowerCAmelCase	)	)
 elif isinstance(_lowerCAmelCase				,  (list, tuple)	):
  for t in tree:
   shapes.extend(_fetch_dims(_lowerCAmelCase	)	)
 elif isinstance(_lowerCAmelCase				,  torch.Tensor	):
  shapes.append(tree.shape	)
 else:
  raise ValueError("Not supported"	)

 return shapes



@torch.jit.ignore
def 						A_							(	_lowerCAmelCase				,  _lowerCAmelCase	)		->					Tuple[int, ...]:
 UpperCamelCase      :   int			      =  []
 for d in reversed(_lowerCAmelCase	):
  idx.append(flat_idx % d	)
  UpperCamelCase      :   List[Any]			      =  flat_idx // d

 return tuple(reversed(_lowerCAmelCase	)	)



@torch.jit.ignore
def 						A_							(	_lowerCAmelCase				,  _lowerCAmelCase				,  _lowerCAmelCase				,  _lowerCAmelCase = None				,  _lowerCAmelCase = None				,  )		->					List[Tuple[slice, ...]]:

 # start_edges and end_edges both indicate whether, starting from any given
 # dimension, the start/end index is at the top/bottom edge of the
 # corresponding tensor, modeled as a tree
 def reduce_edge_list(_lowerCAmelCase	) -> None:
  UpperCamelCase      :   Optional[Any]			      =  True
  for i in range(len(_lowerCAmelCase	)	):
   UpperCamelCase      :   str			      =  -1 * (i + 1)
   l[reversed_idx] &= tally
   UpperCamelCase      :   List[Any]			      =  l[reversed_idx]

 if start_edges is None:
  UpperCamelCase      :   Optional[int]			      =  [s == 0 for s in start]
  reduce_edge_list(_lowerCAmelCase	)
 if end_edges is None:
  UpperCamelCase      :   List[str]			      =  [e == (d - 1) for e, d in zip(_lowerCAmelCase				,  _lowerCAmelCase	)]
  reduce_edge_list(_lowerCAmelCase	)

 # Base cases. Either start/end are empty and we're done, or the final,
 # one-dimensional tensor can be simply sliced
 if len(_lowerCAmelCase	) == 0:
  return [()]
 elif len(_lowerCAmelCase	) == 1:
  return [(slice(start[0]				,  end[0] + 1	),)]

 UpperCamelCase      :   List[Tuple[slice, ...]]			      =  []
 UpperCamelCase      :   List[slice]			      =  []

 # Dimensions common to start and end can be selected directly
 for s, e in zip(_lowerCAmelCase				,  _lowerCAmelCase	):
  if s == e:
   path_list.append(slice(_lowerCAmelCase				,  s + 1	)	)
  else:
   break

 UpperCamelCase      :   Tuple[slice, ...]			      =  tuple(_lowerCAmelCase	)
 UpperCamelCase      :   List[Any]			      =  len(_lowerCAmelCase	)

 # start == end, and we're done
 if divergence_idx == len(_lowerCAmelCase	):
  return [path]

 def upper() -> Tuple[Tuple[slice, ...], ...]:
  assert start_edges is not None
  assert end_edges is not None

  UpperCamelCase      :   Optional[Any]			      =  start[divergence_idx]
  return tuple(
      path + (slice(_lowerCAmelCase				,  sdi + 1	),) + s
      for s in _get_minimal_slice_set(
          start[divergence_idx + 1 :]				,  [d - 1 for d in dims[divergence_idx + 1 :]]				,  dims[divergence_idx + 1 :]				,  start_edges=start_edges[divergence_idx + 1 :]				,  end_edges=[True for _ in end_edges[divergence_idx + 1 :]]				,  )	)

 def lower() -> Tuple[Tuple[slice, ...], ...]:
  assert start_edges is not None
  assert end_edges is not None

  UpperCamelCase      :   int			      =  end[divergence_idx]
  return tuple(
      path + (slice(_lowerCAmelCase				,  edi + 1	),) + s
      for s in _get_minimal_slice_set(
          [0 for _ in start[divergence_idx + 1 :]]				,  end[divergence_idx + 1 :]				,  dims[divergence_idx + 1 :]				,  start_edges=[True for _ in start_edges[divergence_idx + 1 :]]				,  end_edges=end_edges[divergence_idx + 1 :]				,  )	)

 # If both start and end are at the edges of the subtree rooted at
 # divergence_idx, we can just select the whole subtree at once
 if start_edges[divergence_idx] and end_edges[divergence_idx]:
  slices.append(path + (slice(start[divergence_idx]				,  end[divergence_idx] + 1	),)	)
 # If just start is at the edge, we can grab almost all of the subtree,
 # treating only the ragged bottom edge as an edge case
 elif start_edges[divergence_idx]:
  slices.append(path + (slice(start[divergence_idx]				,  end[divergence_idx]	),)	)
  slices.extend(lower()	)
 # Analogous to the previous case, but the top is ragged this time
 elif end_edges[divergence_idx]:
  slices.extend(upper()	)
  slices.append(path + (slice(start[divergence_idx] + 1				,  end[divergence_idx] + 1	),)	)
 # If both sides of the range are ragged, we need to handle both sides
 # separately. If there's contiguous meat in between them, we can index it
 # in one big chunk
 else:
  slices.extend(upper()	)
  UpperCamelCase      :   Optional[int]			      =  end[divergence_idx] - start[divergence_idx]
  if middle_ground > 1:
   slices.append(path + (slice(start[divergence_idx] + 1				,  end[divergence_idx]	),)	)
  slices.extend(lower()	)

 return slices



@torch.jit.ignore
def 						A_							(	_lowerCAmelCase				,  _lowerCAmelCase				,  _lowerCAmelCase				,  _lowerCAmelCase	)		->					torch.Tensor:
 UpperCamelCase      :   Union[str, Any]			      =  t.shape[:no_batch_dims]
 UpperCamelCase      :   str			      =  list(_flat_idx_to_idx(_lowerCAmelCase				,  _lowerCAmelCase	)	)
 # _get_minimal_slice_set is inclusive
 UpperCamelCase      :   List[str]			      =  list(_flat_idx_to_idx(flat_end - 1				,  _lowerCAmelCase	)	)

 # Get an ordered list of slices to perform
 UpperCamelCase      :   Tuple			      =  _get_minimal_slice_set(
     _lowerCAmelCase				,  _lowerCAmelCase				,  _lowerCAmelCase				,  )

 UpperCamelCase      :   int			      =  [t[s] for s in slices]

 return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]	) for s in sliced_tensors]	)



def 						A_							(	_lowerCAmelCase				,  _lowerCAmelCase				,  _lowerCAmelCase				,  _lowerCAmelCase				,  _lowerCAmelCase = False				,  _lowerCAmelCase = None				,  _lowerCAmelCase = False				,  )		->					Any:
 if not (len(_lowerCAmelCase	) > 0):
  raise ValueError("Must provide at least one input"	)

 UpperCamelCase      :   Dict			      =  [shape[:no_batch_dims] for shape in _fetch_dims(_lowerCAmelCase	)]
 UpperCamelCase      :   Union[str, Any]			      =  tuple([max(_lowerCAmelCase	) for s in zip(*_lowerCAmelCase	)]	)

 def _prep_inputs(_lowerCAmelCase	) -> torch.Tensor:
  if not low_mem:
   if not sum(t.shape[:no_batch_dims]	) == no_batch_dims:
    UpperCamelCase      :   str			      =  t.expand(orig_batch_dims + t.shape[no_batch_dims:]	)
   UpperCamelCase      :   Tuple			      =  t.reshape(-1				,  *t.shape[no_batch_dims:]	)
  else:
   UpperCamelCase      :   int			      =  t.expand(orig_batch_dims + t.shape[no_batch_dims:]	)
  return t

 UpperCamelCase      :   Dict[str, Any]			      =  tensor_tree_map(_prep_inputs				,  _lowerCAmelCase	)
 UpperCamelCase      :   Dict			      =  None
 if _out is not None:
  UpperCamelCase      :   Any			      =  tensor_tree_map(lambda _lowerCAmelCase	: t.view([-1] + list(t.shape[no_batch_dims:]	)	)				,  _out	)

 UpperCamelCase      :   Optional[Any]			      =  1
 for d in orig_batch_dims:
  flat_batch_dim *= d

 UpperCamelCase      :   Optional[int]			      =  flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

 def _select_chunk(_lowerCAmelCase	) -> torch.Tensor:
  return t[i : i + chunk_size] if t.shape[0] != 1 else t

 UpperCamelCase      :   Dict			      =  0
 UpperCamelCase      :   Any			      =  prepped_outputs
 for _ in range(_lowerCAmelCase	):
  # Chunk the input
  if not low_mem:
   UpperCamelCase      :   Tuple			      =  _select_chunk
  else:
   UpperCamelCase      :   int			      =  partial(
       _chunk_slice				,  flat_start=_lowerCAmelCase				,  flat_end=min(_lowerCAmelCase				,  i + chunk_size	)				,  no_batch_dims=len(_lowerCAmelCase	)				,  )

  UpperCamelCase      :   Dict[str, Any]			      =  tensor_tree_map(_lowerCAmelCase				,  _lowerCAmelCase	)

  # Run the layer on the chunk
  UpperCamelCase      :   int			      =  layer(**_lowerCAmelCase	)

  # Allocate space for the output
  if out is None:
   UpperCamelCase      :   Union[str, Any]			      =  tensor_tree_map(lambda _lowerCAmelCase	: t.new_zeros((flat_batch_dim,) + t.shape[1:]	)				,  _lowerCAmelCase	)

  # Put the chunk in its pre-allocated space
  if isinstance(_lowerCAmelCase				,  _lowerCAmelCase	):

   def assign(_lowerCAmelCase				,  _lowerCAmelCase	) -> None:
    for k, v in da.items():
     if isinstance(_lowerCAmelCase				,  _lowerCAmelCase	):
      assign(_lowerCAmelCase				,  da[k]	)
     else:
      if _add_into_out:
       v[i : i + chunk_size] += da[k]
      else:
       UpperCamelCase      :   Tuple			      =  da[k]

   assign(_lowerCAmelCase				,  _lowerCAmelCase	)
  elif isinstance(_lowerCAmelCase				,  _lowerCAmelCase	):
   for xa, xa in zip(_lowerCAmelCase				,  _lowerCAmelCase	):
    if _add_into_out:
     xa[i : i + chunk_size] += xa
    else:
     UpperCamelCase      :   Optional[int]			      =  xa
  elif isinstance(_lowerCAmelCase				,  torch.Tensor	):
   if _add_into_out:
    out[i : i + chunk_size] += output_chunk
   else:
    UpperCamelCase      :   Tuple			      =  output_chunk
  else:
   raise ValueError("Not supported"	)

  i += chunk_size

 UpperCamelCase      :   int			      =  tensor_tree_map(lambda _lowerCAmelCase	: t.view(orig_batch_dims + t.shape[1:]	)				,  _lowerCAmelCase	)

 return out



class    A__  :





      def __init__(					self			,				A_ = 512			,				):
       '''simple docstring'''




       UpperCamelCase      :   Optional[int]			      =  max_chunk_size
       UpperCamelCase      :   Optional[int]			      =  None
       UpperCamelCase      :   Optional[tuple]			      =  None





      def 			__UpperCamelCase(					self			,				A_			,				A_			,				A_  ):
       '''simple docstring'''




       logging.info("Tuning chunk size..."  )

       if min_chunk_size >= self.max_chunk_size:
        return min_chunk_size

       UpperCamelCase      :   List[int]			      =  [2**l for l in range(int(math.log(self.max_chunk_size			,				2  )  ) + 1  )]
       UpperCamelCase      :   List[Any]			      =  [c for c in candidates if c > min_chunk_size]
       UpperCamelCase      :   Optional[int]			      =  [min_chunk_size] + candidates
       candidates[-1] += 4

       def test_chunk_size(A_  ) -> bool:
        try:
         with torch.no_grad():
          fn(*A_			,				chunk_size=A_  )
         return True
        except RuntimeError:
         return False

       UpperCamelCase      :   Any			      =  0
       UpperCamelCase      :   int			      =  len(A_  ) - 1
       while i > min_viable_chunk_size_index:
        UpperCamelCase      :   str			      =  test_chunk_size(candidates[i]  )
        if not viable:
         UpperCamelCase      :   List[Any]			      =  (min_viable_chunk_size_index + i) // 2
        else:
         UpperCamelCase      :   List[str]			      =  i
         UpperCamelCase      :   int			      =  (i + len(A_  ) - 1) // 2

       return candidates[min_viable_chunk_size_index]





      def 			__UpperCamelCase(					self			,				A_			,				A_  ):
       '''simple docstring'''




       UpperCamelCase      :   str			      =  True
       for aa, aa in zip(A_			,				A_  ):
        assert type(A_  ) == type(A_  )
        if isinstance(A_			,				(list, tuple)  ):
         consistent &= self._compare_arg_caches(A_			,				A_  )
        elif isinstance(A_			,				A_  ):
         UpperCamelCase      :   int			      =  [v for _, v in sorted(aa.items()			,				key=lambda A_  : x[0]  )]
         UpperCamelCase      :   str			      =  [v for _, v in sorted(aa.items()			,				key=lambda A_  : x[0]  )]
         consistent &= self._compare_arg_caches(A_			,				A_  )
        else:
         consistent &= aa == aa

       return consistent





      def 			__UpperCamelCase(					self			,				A_			,				A_			,				A_			,				):
       '''simple docstring'''




       UpperCamelCase      :   Union[str, Any]			      =  True
       UpperCamelCase      :   tuple			      =  tree_map(lambda A_  : a.shape if isinstance(A_			,				torch.Tensor  ) else a			,				A_			,				A_  )
       if self.cached_arg_data is not None:
        # If args have changed shape/value, we need to re-tune
        assert len(self.cached_arg_data  ) == len(A_  )
        UpperCamelCase      :   List[Any]			      =  self._compare_arg_caches(self.cached_arg_data			,				A_  )
       else:
        # Otherwise, we can reuse the precomputed value
        UpperCamelCase      :   Dict			      =  False

       if not consistent:
        UpperCamelCase      :   Dict			      =  self._determine_favorable_chunk_size(
            A_			,				A_			,				A_			,				)
        UpperCamelCase      :   str			      =  arg_data

       assert self.cached_chunk_size is not None

       return self.cached_chunk_size







from typing import List, Optional, Union

import numpy as np
import torch
import torchaudio.compliance.kaldi as ta_kaldi

from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


__lowerCamelCase			:	List[str]					    =	logging.get_logger(__name__)



class    A__  (			__snake_case     ):
      _UpperCAmelCase      :List[str]		       =  ['input_features', 'attention_mask']





      def __init__(					self			,				A_=80			,				A_=1_6000			,				A_=80			,				A_=0.0			,				A_=True			,				A_=True			,				A_=True			,				**A_			,				):
       '''simple docstring'''




       super().__init__(feature_size=A_			,				sampling_rate=A_			,				padding_value=A_			,				**A_  )
       UpperCamelCase      :   List[Any]			      =  num_mel_bins
       UpperCamelCase      :   Any			      =  do_ceptral_normalize
       UpperCamelCase      :   int			      =  normalize_means
       UpperCamelCase      :   Tuple			      =  normalize_vars
       UpperCamelCase      :   List[Any]			      =  True





      def 			__UpperCamelCase(					self			,				A_			,				):
       '''simple docstring'''




       UpperCamelCase      :   Optional[int]			      =  waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
       UpperCamelCase      :   Optional[Any]			      =  torch.from_numpy(A_  ).unsqueeze(0  )
       UpperCamelCase      :   Dict			      =  ta_kaldi.fbank(A_			,				num_mel_bins=self.num_mel_bins			,				sample_frequency=self.sampling_rate  )
       return features.numpy()





      @staticmethod
      def 			__UpperCamelCase(					A_			,				A_			,				A_ = True			,				A_ = True			,				A_ = 0.0			,				):
       '''simple docstring'''




       if normalize_means:
        UpperCamelCase      :   str			      =  x[:input_length].mean(axis=0  )
        UpperCamelCase      :   Any			      =  np.subtract(A_			,				A_  )
       if normalize_vars:
        UpperCamelCase      :   List[str]			      =  x[:input_length].std(axis=0  )
        UpperCamelCase      :   Optional[int]			      =  np.divide(A_			,				A_  )

       if input_length < x.shape[0]:
        UpperCamelCase      :   Dict			      =  padding_value

       # make sure array is in float32
       UpperCamelCase      :   Union[str, Any]			      =  x.astype(np.floataa  )

       return x





      def 			__UpperCamelCase(					self			,				A_			,				A_ = None  ):
       '''simple docstring'''




       UpperCamelCase      :   Optional[Any]			      =  attention_mask.sum(-1  ) if attention_mask is not None else [x.shape[0] for x in input_features]
       return [
           self.utterance_cmvn(A_			,				A_			,				self.normalize_means			,				self.normalize_vars			,				self.padding_value  )
           for x, n in zip(A_			,				A_  )
       ]





      def __call__(					self			,				A_			,				A_ = False			,				A_ = None			,				A_ = False			,				A_ = None			,				A_ = None			,				A_ = None			,				A_ = None			,				**A_			,				):
       '''simple docstring'''




       if sampling_rate is not None:
        if sampling_rate != self.sampling_rate:
         raise ValueError(
             F"""The model corresponding to this feature extractor: {self} was trained using a sampling rate of"""
             F""" {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled with"""
             F""" {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."  )

       UpperCamelCase      :   str			      =  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}"""  )
       UpperCamelCase      :   List[Any]			      =  is_batched_numpy or (
           isinstance(A_			,				(list, tuple)  ) and (isinstance(raw_speech[0]			,				(np.ndarray, tuple, list)  ))
       )

       if is_batched:
        UpperCamelCase      :   List[str]			      =  [np.asarray(A_			,				dtype=np.floataa  ) for speech in raw_speech]
       elif not is_batched and not isinstance(A_			,				np.ndarray  ):
        UpperCamelCase      :   int			      =  np.asarray(A_			,				dtype=np.floataa  )
       elif isinstance(A_			,				np.ndarray  ) and raw_speech.dtype is np.dtype(np.floataa  ):
        UpperCamelCase      :   Union[str, Any]			      =  raw_speech.astype(np.floataa  )

       # always return batch
       if not is_batched:
        UpperCamelCase      :   Optional[int]			      =  [raw_speech]

       # extract fbank features
       UpperCamelCase      :   Optional[int]			      =  [self._extract_fbank_features(A_  ) for waveform in raw_speech]

       # convert into correct format for padding
       UpperCamelCase      :   List[Any]			      =  BatchFeature({"input_features": features}  )

       UpperCamelCase      :   Any			      =  self.pad(
           A_			,				padding=A_			,				max_length=A_			,				truncation=A_			,				pad_to_multiple_of=A_			,				return_attention_mask=A_			,				**A_			,				)

       # make sure list is in array format
       UpperCamelCase      :   Optional[int]			      =  padded_inputs.get("input_features"  )
       if isinstance(input_features[0]			,				A_  ):
        UpperCamelCase      :   str			      =  [np.asarray(A_			,				dtype=np.floataa  ) for feature in input_features]

       UpperCamelCase      :   Optional[int]			      =  padded_inputs.get("attention_mask"  )
       if attention_mask is not None:
        UpperCamelCase      :   Optional[int]			      =  [np.asarray(A_			,				dtype=np.intaa  ) for array in attention_mask]

       # Utterance-level cepstral mean and variance normalization
       if self.do_ceptral_normalize:
        UpperCamelCase      :   List[Any]			      =  (
            np.array(A_			,				dtype=np.intaa  )
            if self._get_padding_strategies(A_			,				max_length=A_  ) is not PaddingStrategy.DO_NOT_PAD
            else None
        )
        UpperCamelCase      :   int			      =  self.normalize(
            padded_inputs["input_features"]			,				attention_mask=A_  )

       if return_tensors is not None:
        UpperCamelCase      :   List[str]			      =  padded_inputs.convert_to_tensors(A_  )

       return padded_inputs
