




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


_lowerCamelCase     :	List[Any]         =				logging.get_logger(__name__)

class 				lowercase					(			UpperCamelCase__):
				__lowerCAmelCase			:			Optional[Any]     				=      ["""input_features"""]







				def __init__(							self      :    Union[str, Any]					,   _lowerCamelCase      :    int=80					,   _lowerCamelCase      :    List[Any]=1_60_00					,   _lowerCamelCase      :    Optional[Any]=1_60					,   _lowerCamelCase      :    str=30					,   _lowerCamelCase      :    str=4_00					,   _lowerCamelCase      :    int=0.0					,   _lowerCamelCase      :    Any=False					,   **_lowerCamelCase      :    List[str]					,   ):



					"""simple docstring"""

					super().__init__(
					    feature_size=lowerCAmelCase__					,   sampling_rate=lowerCAmelCase__					,   padding_value=lowerCAmelCase__					,   return_attention_mask=lowerCAmelCase__					,   **lowerCAmelCase__					,   )
					A_     : Any		      =						n_fft
					A_     : List[Any]		      =						hop_length
					A_     : Any		      =						chunk_length
					A_     : List[Any]		      =						chunk_length * sampling_rate
					A_     : Tuple		      =						self.n_samples // hop_length
					A_     : str		      =						sampling_rate
					A_     : Dict		      =						mel_filter_bank(
					    num_frequency_bins=1 + n_fft // 2					,   num_mel_filters=lowerCAmelCase__					,   min_frequency=0.0					,   max_frequency=80_00.0					,   sampling_rate=lowerCAmelCase__					,   norm='''slaney'''					,   mel_scale='''slaney'''					,   )







				def     a_     (							self      :    Any					,   _lowerCamelCase      :    Any				):



					"""simple docstring"""

					A_     : Tuple		      =						spectrogram(
					    lowerCAmelCase__					,   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_     : List[Any]		      =						log_spec[:, :-1]
					A_     : Optional[int]		      =						np.maximum(lowerCAmelCase__					,   log_spec.max() - 8.0				)
					A_     : Optional[Any]		      =						(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     a_     (							_lowerCamelCase      :    List[str]					,   _lowerCamelCase      :    Optional[int]					,   _lowerCamelCase      :    Dict = 0.0				):



					"""simple docstring"""

					if attention_mask is not None:
						A_     : Union[str, Any]		      =						np.array(lowerCAmelCase__					,   np.intaa				)
						A_     : str		      =						[]

						for vector, length in zip(lowerCAmelCase__					,   attention_mask.sum(-1				)				):
							A_     : Union[str, Any]		      =						(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7				)
							if length < normed_slice.shape[0]:
								A_     : Union[str, Any]		      =						padding_value

							normed_input_values.append(lowerCAmelCase__				)
					else:
						A_     : Tuple		      =						[(x - x.mean()) / np.sqrt(x.var() + 1E-7				) for x in input_values]

					return normed_input_values







				def __call__(							self      :    Tuple					,   _lowerCamelCase      :    Optional[int]					,   _lowerCamelCase      :    Optional[Any] = True					,   _lowerCamelCase      :    List[str] = None					,   _lowerCamelCase      :    Tuple = None					,   _lowerCamelCase      :    Optional[int] = None					,   _lowerCamelCase      :    Optional[Any] = "max_length"					,   _lowerCamelCase      :    int = None					,   _lowerCamelCase      :    Any = None					,   _lowerCamelCase      :    Any = None					,   **_lowerCamelCase      :    Union[str, Any]					,   ):



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

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

					# always return batch
					if not is_batched:
						A_     : Union[str, Any]		      =						[np.asarray([raw_speech]				).T]

					A_     : Optional[Any]		      =						BatchFeature({'''input_features''': raw_speech}				)

					# convert into correct format for padding

					A_     : List[str]		      =						self.pad(
					    lowerCAmelCase__					,   padding=lowerCAmelCase__					,   max_length=max_length if max_length else self.n_samples					,   truncation=lowerCAmelCase__					,   pad_to_multiple_of=lowerCAmelCase__					,   return_attention_mask=return_attention_mask or do_normalize					,   )

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

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

					A_     : Optional[Any]		      =						[self._np_extract_fbank_features(lowerCAmelCase__				) for waveform in input_features[0]]

					if isinstance(input_features[0]					,   lowerCAmelCase__				):
						A_     : Dict		      =						[np.asarray(lowerCAmelCase__					,   dtype=np.floataa				) for feature in input_features]
					else:
						A_     : int		      =						input_features

					if return_attention_mask:
						# rescale from sample (48000) to feature (3000)
						A_     : Optional[int]		      =						padded_inputs["attention_mask"][:, :: self.hop_length]

					if return_tensors is not None:
						A_     : Optional[int]		      =						padded_inputs.convert_to_tensors(lowerCAmelCase__				)

					return padded_inputs







				def     a_     (							self      :    List[Any]				):



					"""simple docstring"""

					A_     : List[Any]		      =						copy.deepcopy(self.__dict__				)
					A_     : Optional[int]		      =						self.__class__.__name__
					if "mel_filters" in output:
						del output["mel_filters"]
					return output












from queue import Queue
from typing import TYPE_CHECKING, Optional


if TYPE_CHECKING:
     from ..models.auto import AutoTokenizer







class 				__lowerCAmelCase       :







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						Optional[Any]:



        '''simple docstring'''




        raise NotImplementedError()







      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        raise NotImplementedError()







class 				__lowerCAmelCase       (    UpperCamelCase__):







      def __init__(    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = False		,      **lowerCAmelCase__							)   ->						int:



        '''simple docstring'''




        a__	:					str										=tokenizer
        a__	:					List[str]										=skip_prompt
        a__	:					List[Any]										=decode_kwargs

        # variables used in the streaming process
        a__	:					Dict										=[]
        a__	:					int										=0
        a__	:					str										=True







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						Union[str, Any]:



        '''simple docstring'''




        if len(value.shape							) > 1 and value.shape[0] > 1:
          raise ValueError("TextStreamer only supports batch size 1"							)
        elif len(value.shape							) > 1:
          a__	:					Any										=value[0]

        if self.skip_prompt and self.next_tokens_are_prompt:
          a__	:					Dict										=False
          return

        # Add the new token to the cache and decodes the entire thing.
        self.token_cache.extend(value.tolist()							)
        a__	:					Union[str, Any]										=self.tokenizer.decode(self.token_cache		,      **self.decode_kwargs							)

        # After the symbol for a new line, we flush the cache.
        if text.endswith("\n"							):
          a__	:					List[Any]										=text[self.print_len :]
          a__	:					List[str]										=[]
          a__	:					Optional[int]										=0
        # If the last token is a CJK character, we print the characters.
        elif len(lowerCAmelCase__							) > 0 and self._is_chinese_char(ord(text[-1]							)							):
          a__	:					List[str]										=text[self.print_len :]
          self.print_len += len(lowerCAmelCase__							)
        # Otherwise, prints until the last space char (simple heuristic to avoid printing incomplete words,
        # which may change with the subsequent token -- there are probably smarter ways to do this!)
        else:
          a__	:					str										=text[self.print_len : text.rfind(" "							) + 1]
          self.print_len += len(lowerCAmelCase__							)

        self.on_finalized_text(lowerCAmelCase__							)







      def    _lowercase  (    self							)   ->						Any:



        '''simple docstring'''




        if len(self.token_cache							) > 0:
          a__	:					Union[str, Any]										=self.tokenizer.decode(self.token_cache		,      **self.decode_kwargs							)
          a__	:					List[Any]										=text[self.print_len :]
          a__	:					List[str]										=[]
          a__	:					Optional[int]										=0
        else:
          a__	:					Union[str, Any]										=""

        a__	:					Any										=True
        self.on_finalized_text(lowerCAmelCase__		,      stream_end=lowerCAmelCase__							)







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = False							)   ->						Optional[Any]:



        '''simple docstring'''




        print(lowerCAmelCase__		,      flush=lowerCAmelCase__		,      end="" if not stream_end else None							)







      def    _lowercase  (    self		,      lowerCAmelCase__							)   ->						str:



        '''simple docstring'''




        if (
            (cp >= 0X4E_00 and cp <= 0X9F_FF)
            or (cp >= 0X34_00 and cp <= 0X4D_BF)  #
            or (cp >= 0X2_00_00 and cp <= 0X2_A6_DF)  #
            or (cp >= 0X2_A7_00 and cp <= 0X2_B7_3F)  #
            or (cp >= 0X2_B7_40 and cp <= 0X2_B8_1F)  #
            or (cp >= 0X2_B8_20 and cp <= 0X2_CE_AF)  #
            or (cp >= 0XF9_00 and cp <= 0XFA_FF)
            or (cp >= 0X2_F8_00 and cp <= 0X2_FA_1F)  #
        ):  #
          return True

        return False







class 				__lowerCAmelCase       (    UpperCamelCase__):







      def __init__(    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = False		,      lowerCAmelCase__ = None		,      **lowerCAmelCase__							)   ->						Optional[Any]:



        '''simple docstring'''




        super().__init__(lowerCAmelCase__		,      lowerCAmelCase__		,      **lowerCAmelCase__							)
        a__	:					str										=Queue()
        a__	:					Optional[Any]										=None
        a__	:					Any										=timeout







      def    _lowercase  (    self		,      lowerCAmelCase__		,      lowerCAmelCase__ = False							)   ->						List[str]:



        '''simple docstring'''




        self.text_queue.put(lowerCAmelCase__		,      timeout=self.timeout							)
        if stream_end:
          self.text_queue.put(self.stop_signal		,      timeout=self.timeout							)







      def __iter__(    self							)   ->						Dict:



        '''simple docstring'''




        return self







      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        a__	:					int										=self.text_queue.get(timeout=self.timeout							)
        if value == self.stop_signal:
          raise StopIteration()
        else:
          return value






