




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


UpperCAmelCase_									=	logging.get_logger(__name__)





class     lowerCamelCase__(					__lowercase):
      UpperCAmelCase__      :	str    		=   ['input_features', 'attention_mask']
      def __init__(				self: List[Any]    ,       UpperCamelCase_: int=80    ,       UpperCamelCase_: int=1_60_00    ,       UpperCamelCase_: Any=80    ,       UpperCamelCase_: str=0.0    ,       UpperCamelCase_: Any=True    ,       UpperCamelCase_: List[Any]=True    ,       UpperCamelCase_: Optional[int]=True    ,       **UpperCamelCase_: int    ,       ):
             super().__init__(feature_size=UpperCamelCase_    ,       sampling_rate=UpperCamelCase_    ,       padding_value=UpperCamelCase_    ,       **UpperCamelCase_   )
             __lowerCamelCase   =						num_mel_bins
             __lowerCamelCase   =						do_ceptral_normalize
             __lowerCamelCase   =						normalize_means
             __lowerCamelCase   =						normalize_vars
             __lowerCamelCase   =						True
      def 			lowerCAmelCase__						(				self: List[Any]    ,       UpperCamelCase_: np.ndarray    ,       ):
             __lowerCamelCase   =						waveform * (2**15)  # Kaldi compliance: 16-bit signed integers
             __lowerCamelCase   =						torch.from_numpy(UpperCamelCase_   ).unsqueeze(0   )
             __lowerCamelCase   =						ta_kaldi.fbank(UpperCamelCase_    ,       num_mel_bins=self.num_mel_bins    ,       sample_frequency=self.sampling_rate   )
             return features.numpy()
      @staticmethod
      def 			lowerCAmelCase__						(				UpperCamelCase_: np.ndarray    ,       UpperCamelCase_: int    ,       UpperCamelCase_: Optional[bool] = True    ,       UpperCamelCase_: Optional[bool] = True    ,       UpperCamelCase_: float = 0.0    ,       ):
             # make sure we normalize float32 arrays
             if normalize_means:
                    __lowerCamelCase   =						x[:input_length].mean(axis=0   )
                    __lowerCamelCase   =						np.subtract(UpperCamelCase_    ,       UpperCamelCase_   )
             if normalize_vars:
                    __lowerCamelCase   =						x[:input_length].std(axis=0   )
                    __lowerCamelCase   =						np.divide(UpperCamelCase_    ,       UpperCamelCase_   )

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

             # make sure array is in float32
             __lowerCamelCase   =						x.astype(np.floataa   )

             return x
      def 			lowerCAmelCase__						(				self: Any    ,       UpperCamelCase_: List[np.ndarray]    ,       UpperCamelCase_: Optional[np.ndarray] = None   ):
             __lowerCamelCase   =						attention_mask.sum(-1   ) if attention_mask is not None else [x.shape[0] for x in input_features]
             return [
                 self.utterance_cmvn(UpperCamelCase_    ,       UpperCamelCase_    ,       self.normalize_means    ,       self.normalize_vars    ,       self.padding_value   )
                 for x, n in zip(UpperCamelCase_    ,       UpperCamelCase_   )
             ]




      def __call__(				self: List[str]    ,       UpperCamelCase_: Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]    ,       UpperCamelCase_: Union[bool, str, PaddingStrategy] = False    ,       UpperCamelCase_: Optional[int] = None    ,       UpperCamelCase_: bool = False    ,       UpperCamelCase_: Optional[int] = None    ,       UpperCamelCase_: Optional[Union[str, TensorType]] = None    ,       UpperCamelCase_: Optional[int] = None    ,       UpperCamelCase_: Optional[bool] = None    ,       **UpperCamelCase_: List[str]    ,       ):

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

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

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

             # always return batch
             if not is_batched:
                    __lowerCamelCase   =						[raw_speech]

             # extract fbank features
             __lowerCamelCase   =						[self._extract_fbank_features(UpperCamelCase_   ) for waveform in raw_speech]

             # convert into correct format for padding
             __lowerCamelCase   =						BatchFeature({"""input_features""": features}   )

             __lowerCamelCase   =						self.pad(
                 UpperCamelCase_    ,       padding=UpperCamelCase_    ,       max_length=UpperCamelCase_    ,       truncation=UpperCamelCase_    ,       pad_to_multiple_of=UpperCamelCase_    ,       return_attention_mask=UpperCamelCase_    ,       **UpperCamelCase_    ,       )

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

             __lowerCamelCase   =						padded_inputs.get("""attention_mask"""   )
             if attention_mask is not None:
                    __lowerCamelCase   =						[np.asarray(UpperCamelCase_    ,       dtype=np.intaa   ) for array in attention_mask]

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

             if return_tensors is not None:
                    __lowerCamelCase   =						padded_inputs.convert_to_tensors(UpperCamelCase_   )

             return padded_inputs



import PIL.Image
import PIL.ImageOps
from packaging import version
from PIL import Image


if version.parse(version.parse(PIL.__version__).base_version) >= version.parse('9.1.0'):
		lowercase  :					int													=       {
		    'linear': PIL.Image.Resampling.BILINEAR,
		    'bilinear': PIL.Image.Resampling.BILINEAR,
		    'bicubic': PIL.Image.Resampling.BICUBIC,
		    'lanczos': PIL.Image.Resampling.LANCZOS,
		    'nearest': PIL.Image.Resampling.NEAREST,
		}
else:
		lowercase  :					Any													=       {
		    'linear': PIL.Image.LINEAR,
		    'bilinear': PIL.Image.BILINEAR,
		    'bicubic': PIL.Image.BICUBIC,
		    'lanczos': PIL.Image.LANCZOS,
		    'nearest': PIL.Image.NEAREST,
		}




def 	_SCREAMING_SNAKE_CASE					(		_lowerCamelCase :			int)					->    Any:






							'''simple docstring'''





							__UpperCamelCase	:   Union[str, Any]			   =					(images / 2 + 0.5).clamp(0						,  1)
							__UpperCamelCase	:   int			   =					images.cpu().permute(0						,  2						,  3						,  1).float().numpy()
							__UpperCamelCase	:   Dict			   =					numpy_to_pil(_lowerCamelCase)
							return images




def 	_SCREAMING_SNAKE_CASE					(		_lowerCamelCase :			Optional[Any])					->    Optional[Any]:






							'''simple docstring'''





							if images.ndim == 3:
														__UpperCamelCase	:   Optional[int]			   =					images[None, ...]
							__UpperCamelCase	:   Dict			   =					(images * 255).round().astype("uint8")
							if images.shape[-1] == 1:
														# special case for grayscale (single channel) images
														__UpperCamelCase	:   int			   =					[Image.fromarray(image.squeeze()						,  mode="L") for image in images]
							else:
														__UpperCamelCase	:   Union[str, Any]			   =					[Image.fromarray(_lowerCamelCase) for image in images]

							return pil_images