






from math import ceil
from typing import List, Optional, Union

import numpy as np

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


__a   :							Union[str, Any]						= logging.get_logger(__name__)


class    __lowercase   (						a__	):







     '''simple docstring'''
     SCREAMING_SNAKE_CASE     =					["audio_values", "audio_mask"]




     def __init__(					self							:      Any			,						UpperCamelCase_							:      List[Any]=2_048			,						UpperCamelCase_							:      int=1			,						UpperCamelCase_							:      int=[16, 16]			,						UpperCamelCase_							:      List[str]=128			,						UpperCamelCase_							:      Tuple=44_100			,						UpperCamelCase_							:      Tuple=86			,						UpperCamelCase_							:      Dict=2_048			,						UpperCamelCase_							:      List[str]=0.0			,						**UpperCamelCase_							:      List[str]			,						):


         """simple docstring"""



         super().__init__(
             feature_size=__a			,						sampling_rate=__a			,						padding_value=__a			,						**__a			,						)

         __A       =						spectrogram_length
         __A       =						num_channels
         __A       =						patch_size
         __A       =						feature_size // self.patch_size[1]
         __A       =						n_fft
         __A       =						sampling_rate // hop_length_to_sampling_rate
         __A       =						sampling_rate
         __A       =						padding_value
         __A       =						mel_filter_bank(
             num_frequency_bins=1 + n_fft // 2			,						num_mel_filters=__a			,						min_frequency=0.0			,						max_frequency=22_050.0			,						sampling_rate=__a			,						norm="""slaney"""			,						mel_scale="""slaney"""			,						).T




     def 		lowerCAmelCase_  (					self							:      Union[str, Any]			,						UpperCamelCase_							:      np.array					):


         """simple docstring"""



         __A       =						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.T			,						log_mel="""dB"""			,						db_range=80.0			,						)
         __A       =						log_spec[:, :-1]
         __A       =						log_spec - 20.0
         __A       =						np.clip(log_spec / 40.0			,						-2.0			,						0.0					) + 1.0
         return log_spec





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


         """simple docstring"""



         if sampling_rate is not None:
             if sampling_rate != self.sampling_rate:
                 raise ValueError(
                     """This feature extractor is set to support sampling rate"""
                     F" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"
                     F" 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(__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}"					)
         __A       =						is_batched_numpy or (
             isinstance(__a			,						(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(__a			,						np.ndarray					):
             __A       =						np.asarray(__a			,						dtype=np.floataa					)
         elif isinstance(__a			,						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]

         # Convert audio signals to log mel spectrograms, truncate by time axis
         __A       =						[
             self._np_extract_fbank_features(waveform.squeeze()					).T[: self.spectrogram_length] for waveform in raw_speech
         ]
         if isinstance(audio_features[0]			,						__a					):
             __A       =						[np.asarray(__a			,						dtype=np.floataa					) for feature in audio_features]

         # Create audio attention mask
         __A       =						max(
             [ceil(feature.shape[0] / self.patch_size[0]					) * self.freq_len for feature in audio_features]					)  # The maximum number of audio patches in a batch
         if return_attention_mask:
             __A       =						[
                 (ceil(feature.shape[0] / self.patch_size[0]					) * self.freq_len) * [1]
                 + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]					) * self.freq_len) * [0]
                 for feature in audio_features
             ]
             __A       =						np.array(__a					).astype(np.floataa					)

         # convert into correct format for padding
         __A       =						max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
         __A       =						np.ones([len(__a					), 1, max_time_len, self.feature_size]					).astype(np.floataa					)
         __A       =						padded_audio_features * self.padding_value
         for i in range(len(__a					)					):
             __A       =						audio_features[i]
             __A       =						feature

         # return as BatchFeature
         if return_attention_mask:
             __A       =						{"""audio_values""": padded_audio_features, """audio_mask""": audio_mask}
         else:
             __A       =						{"""audio_values""": padded_audio_features}

         __A       =						BatchFeature(data=__a			,						tensor_type=__a					)
         return encoded_inputs








# Copyright 2023 The HuggingFace Inc. 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 copy
import importlib.metadata
import json
import os
from dataclasses import dataclass
from typing import Any, Dict, Union

from packaging import version

from ..utils import is_torch_available, logging


if is_torch_available():
      import torch


snake_case__				:	int					       =					logging.get_logger(__name__)





@dataclass
class   SCREAMING_SNAKE_CASE_			:




       '''simple docstring'''



       def __init__(						self				:				int		,				__a				:				Optional[Any]=False		,				__a				:				Tuple=False		,				__a				:				Tuple=6.0		,				__a				:				str=None		,				__a				:				List[str]=False		,				__a				:				str=False		,				__a				:				Optional[Any]=None		,				__a				:				Dict="fp4"		,				__a				:				Optional[int]=False		,				**__a				:				Dict		,				)					->str:
          lowerCamelCase_						:   str		= load_in_abit
          lowerCamelCase_						:   Union[str, Any]		= load_in_abit
          lowerCamelCase_						:   Tuple		= llm_inta_threshold
          lowerCamelCase_						:   int		= llm_inta_skip_modules
          lowerCamelCase_						:   List[str]		= llm_inta_enable_fpaa_cpu_offload
          lowerCamelCase_						:   Optional[Any]		= llm_inta_has_fpaa_weight
          lowerCamelCase_						:   Dict		= bnb_abit_quant_type
          lowerCamelCase_						:   int		= bnb_abit_use_double_quant

          if bnb_abit_compute_dtype is None:
             lowerCamelCase_						:   int		= torch.floataa
          elif isinstance(__a		,				__a							):
             lowerCamelCase_						:   Tuple		= getattr(__a		,				__a							)
          elif isinstance(__a		,				torch.dtype							):
             lowerCamelCase_						:   List[str]		= bnb_abit_compute_dtype
          else:
             raise ValueError("""bnb_4bit_compute_dtype must be a string or a torch.dtype"""							)

          self.post_init()



       def 				_lowerCAmelCase							(						self				:				Any							)					->Union[str, Any]:
          if not isinstance(self.llm_inta_threshold		,				__a							):
             raise ValueError("""llm_int8_threshold must be a float"""							)

          if self.llm_inta_skip_modules is not None and not isinstance(self.llm_inta_skip_modules		,				__a							):
             raise ValueError("""llm_int8_skip_modules must be a list of strings"""							)
          if not isinstance(self.llm_inta_enable_fpaa_cpu_offload		,				__a							):
             raise ValueError("""llm_int8_enable_fp32_cpu_offload must be a boolean"""							)

          if not isinstance(self.llm_inta_has_fpaa_weight		,				__a							):
             raise ValueError("""llm_int8_has_fp16_weight must be a boolean"""							)

          if self.bnb_abit_compute_dtype is not None and not isinstance(self.bnb_abit_compute_dtype		,				torch.dtype							):
             raise ValueError("""bnb_4bit_compute_dtype must be torch.dtype"""							)

          if not isinstance(self.bnb_abit_quant_type		,				__a							):
             raise ValueError("""bnb_4bit_quant_type must be a string"""							)

          if not isinstance(self.bnb_abit_use_double_quant		,				__a							):
             raise ValueError("""bnb_4bit_use_double_quant must be a boolean"""							)

          if self.load_in_abit and not version.parse(importlib.metadata.version("""bitsandbytes"""							)							) >= version.parse(
              """0.39.0"""							):
             raise ValueError(
                 """4 bit quantization requires bitsandbytes>=0.39.0 - please upgrade your bitsandbytes version"""							)



       def 				_lowerCAmelCase							(						self				:				Optional[int]							)					->int:
          return self.load_in_abit or self.load_in_abit



       def 				_lowerCAmelCase							(						self				:				List[str]							)					->List[Any]:
          if self.load_in_abit:
             return "llm_int8"
          elif self.load_in_abit and self.bnb_abit_quant_type == "fp4":
             return "fp4"
          elif self.load_in_abit and self.bnb_abit_quant_type == "nf4":
             return "nf4"
          else:
             return None



       @classmethod
       def 				_lowerCAmelCase							(						cls				:				int		,				__a				:				Dict		,				__a				:				str		,				**__a				:				Any							)					->Optional[Any]:
          lowerCamelCase_						:   Optional[Any]		= cls(**__a							)

          lowerCamelCase_						:   List[Any]		= []
          for key, value in kwargs.items():
             if hasattr(__a		,				__a							):
                setattr(__a		,				__a		,				__a							)
                to_remove.append(__a							)
          for key in to_remove:
             kwargs.pop(__a		,				__a							)

          if return_unused_kwargs:
             return config, kwargs
          else:
             return config



       def 				_lowerCAmelCase							(						self				:				Optional[Any]		,				__a				:				Union[str, os.PathLike]							)					->Any:
          with open(__a		,				"""w"""		,				encoding="""utf-8"""							) as writer:
             lowerCamelCase_						:   List[Any]		= self.to_dict()
             lowerCamelCase_						:   Dict		= json.dumps(__a		,				indent=2		,				sort_keys=__a							) + """\n"""

             writer.write(__a							)



       def 				_lowerCAmelCase							(						self				:				Tuple							)					->Dict[str, Any]:
          lowerCamelCase_						:   Dict		= copy.deepcopy(self.__dict__							)
          lowerCamelCase_						:   Tuple		= str(output["""bnb_4bit_compute_dtype"""]							).split("""."""							)[1]

          return output



       def __repr__(						self				:				List[str]							)					->Any:
          return F'''{self.__class__.__name__} {self.to_json_string()}'''



       def 				_lowerCAmelCase							(						self				:				Optional[int]		,				__a				:				bool = True							)					->str:
          if use_diff is True:
             lowerCamelCase_						:   Dict		= self.to_diff_dict()
          else:
             lowerCamelCase_						:   Any		= self.to_dict()
          return json.dumps(__a		,				indent=2		,				sort_keys=__a							) + "\n"




       def 				_lowerCAmelCase							(						self				:				Optional[Any]							)					->Dict[str, Any]:
          lowerCamelCase_						:   Any		= self.to_dict()

          # get the default config dict
          lowerCamelCase_						:   List[str]		= BitsAndBytesConfig().to_dict()

          lowerCamelCase_						:   str		= {}

          # only serialize values that differ from the default config
          for key, value in config_dict.items():
             if value != default_config_dict[key]:
                lowerCamelCase_						:   List[Any]		= value

          return serializable_config_dict





