





import itertools
import random
import unittest

import numpy as np

from transformers import is_speech_available
from transformers.testing_utils import require_torch, require_torchaudio

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_speech_available():
     from transformers import SpeechaTextFeatureExtractor

UpperCAmelCase	:	Any		     =       random.Random()
def      _A       (		SCREAMING_SNAKE_CASE     :   Dict       ,       SCREAMING_SNAKE_CASE     :   Optional[Any]=1.0       ,       SCREAMING_SNAKE_CASE     :   Optional[int]=None       ,       SCREAMING_SNAKE_CASE     :   Optional[int]=None						):





  """simple docstring"""



  if rng is None:
    a__	:					Optional[int]										=global_rng

  a__	:					Optional[Any]										=[]
  for batch_idx in range(shape[0]						):
    values.append([]						)
    for _ in range(shape[1]						):
      values[-1].append(rng.random() * scale						)

  return values







@require_torch
@require_torchaudio
class 				__lowerCAmelCase       (    unittest.TestCase):







      def __init__(    self		,      lowerCAmelCase__		,      lowerCAmelCase__=7		,      lowerCAmelCase__=4_0_0		,      lowerCAmelCase__=2_0_0_0		,      lowerCAmelCase__=2_4		,      lowerCAmelCase__=2_4		,      lowerCAmelCase__=0.0		,      lowerCAmelCase__=1_6_0_0_0		,      lowerCAmelCase__=True		,      lowerCAmelCase__=True		,      )   ->						Tuple:



        '''simple docstring'''




        a__	:					int										=parent
        a__	:					str										=batch_size
        a__	:					int										=min_seq_length
        a__	:					Any										=max_seq_length
        a__	:					Optional[Any]										=(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        a__	:					Optional[int]										=feature_size
        a__	:					Tuple										=num_mel_bins
        a__	:					Optional[int]										=padding_value
        a__	:					Optional[Any]										=sampling_rate
        a__	:					List[str]										=return_attention_mask
        a__	:					str										=do_normalize







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



        '''simple docstring'''




        return {
            "feature_size": self.feature_size,
            "num_mel_bins": self.num_mel_bins,
            "padding_value": self.padding_value,
            "sampling_rate": self.sampling_rate,
            "return_attention_mask": self.return_attention_mask,
            "do_normalize": self.do_normalize,
        }







      def    _lowercase  (    self		,      lowerCAmelCase__=False		,      lowerCAmelCase__=False							)   ->						Dict:



        '''simple docstring'''




        def _flatten(lowerCAmelCase__							):
          return list(itertools.chain(*lowerCAmelCase__							)							)

        if equal_length:
          a__	:					Any										=[floats_list((self.max_seq_length, self.feature_size)							) for _ in range(self.batch_size							)]
        else:
          # make sure that inputs increase in size
          a__	:					Optional[int]										=[
              floats_list((x, self.feature_size)							)
              for x in range(self.min_seq_length		,      self.max_seq_length		,      self.seq_length_diff							)
          ]
        if numpify:
          a__	:					Optional[int]										=[np.asarray(lowerCAmelCase__							) for x in speech_inputs]
        return speech_inputs







@require_torch
@require_torchaudio
class 				__lowerCAmelCase       (    UpperCamelCase__						,				unittest.TestCase):
      _lowercase				:					List[Any]						=			SpeechaTextFeatureExtractor if is_speech_available() else None







      def    _lowercase  (    self							)   ->						List[Any]:



        '''simple docstring'''




        a__	:					Any										=SpeechaTextFeatureExtractionTester(self							)







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



        '''simple docstring'''




        self.assertTrue(np.all(np.mean(lowerCAmelCase__		,      axis=0							) < 1E-3							)							)
        self.assertTrue(np.all(np.abs(np.var(lowerCAmelCase__		,      axis=0							) - 1							) < 1E-3							)							)







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



        '''simple docstring'''




        a__	:					List[str]										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        # create three inputs of length 800, 1000, and 1200
        a__	:					List[Any]										=[floats_list((1, x)							)[0] for x in range(8_0_0		,      1_4_0_0		,      2_0_0							)]
        a__	:					List[Any]										=[np.asarray(lowerCAmelCase__							) for speech_input in speech_inputs]

        # Test feature size
        a__	:					int										=feature_extractor(lowerCAmelCase__		,      padding=lowerCAmelCase__		,      return_tensors="np"							).input_features
        self.assertTrue(input_features.ndim == 3							)
        self.assertTrue(input_features.shape[-1] == feature_extractor.feature_size							)

        # Test not batched input
        a__	:					Optional[int]										=feature_extractor(speech_inputs[0]		,      return_tensors="np"							).input_features
        a__	:					Optional[Any]										=feature_extractor(np_speech_inputs[0]		,      return_tensors="np"							).input_features
        self.assertTrue(np.allclose(lowerCAmelCase__		,      lowerCAmelCase__		,      atol=1E-3							)							)

        # Test batched
        a__	:					Dict										=feature_extractor(lowerCAmelCase__		,      return_tensors="np"							).input_features
        a__	:					Union[str, Any]										=feature_extractor(lowerCAmelCase__		,      return_tensors="np"							).input_features
        for enc_seq_a, enc_seq_a in zip(lowerCAmelCase__		,      lowerCAmelCase__							):
          self.assertTrue(np.allclose(lowerCAmelCase__		,      lowerCAmelCase__		,      atol=1E-3							)							)

        # Test 2-D numpy arrays are batched.
        a__	:					Optional[int]										=[floats_list((1, x)							)[0] for x in (8_0_0, 8_0_0, 8_0_0)]
        a__	:					Any										=np.asarray(lowerCAmelCase__							)
        a__	:					List[str]										=feature_extractor(lowerCAmelCase__		,      return_tensors="np"							).input_features
        a__	:					str										=feature_extractor(lowerCAmelCase__		,      return_tensors="np"							).input_features
        for enc_seq_a, enc_seq_a in zip(lowerCAmelCase__		,      lowerCAmelCase__							):
          self.assertTrue(np.allclose(lowerCAmelCase__		,      lowerCAmelCase__		,      atol=1E-3							)							)







      def    _lowercase  (    self							)   ->						Any:



        '''simple docstring'''




        a__	:					List[str]										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        a__	:					Dict										=[floats_list((1, x)							)[0] for x in range(8_0_0		,      1_4_0_0		,      2_0_0							)]

        a__	:					int										=["longest", "max_length", "do_not_pad"]
        a__	:					Optional[Any]										=[None, 1_6, None]
        for max_length, padding in zip(lowerCAmelCase__		,      lowerCAmelCase__							):
          a__	:					List[str]										=feature_extractor(
              lowerCAmelCase__		,      padding=lowerCAmelCase__		,      max_length=lowerCAmelCase__		,      return_attention_mask=lowerCAmelCase__							)
          a__	:					Dict										=inputs.input_features
          a__	:					Optional[int]										=inputs.attention_mask
          a__	:					List[Any]										=[np.sum(lowerCAmelCase__							) for x in attention_mask]

          self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]							)
          self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]							)
          self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]							)







      def    _lowercase  (    self							)   ->						Any:



        '''simple docstring'''




        a__	:					List[str]										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        a__	:					List[str]										=[floats_list((1, x)							)[0] for x in range(8_0_0		,      1_4_0_0		,      2_0_0							)]

        a__	:					Union[str, Any]										=["longest", "max_length", "do_not_pad"]
        a__	:					Optional[int]										=[None, 1_6, None]
        for max_length, padding in zip(lowerCAmelCase__		,      lowerCAmelCase__							):
          a__	:					Optional[int]										=feature_extractor(
              lowerCAmelCase__		,      max_length=lowerCAmelCase__		,      padding=lowerCAmelCase__		,      return_tensors="np"		,      return_attention_mask=lowerCAmelCase__							)
          a__	:					Optional[int]										=inputs.input_features
          a__	:					List[str]										=inputs.attention_mask
          a__	:					Union[str, Any]										=[np.sum(lowerCAmelCase__							) for x in attention_mask]

          self._check_zero_mean_unit_variance(input_features[0][: fbank_feat_lengths[0]]							)
          self.assertTrue(input_features[0][fbank_feat_lengths[0] :].sum() < 1E-6							)
          self._check_zero_mean_unit_variance(input_features[1][: fbank_feat_lengths[1]]							)
          self.assertTrue(input_features[0][fbank_feat_lengths[1] :].sum() < 1E-6							)
          self._check_zero_mean_unit_variance(input_features[2][: fbank_feat_lengths[2]]							)







      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        a__	:					str										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        a__	:					List[Any]										=[floats_list((1, x)							)[0] for x in range(8_0_0		,      1_4_0_0		,      2_0_0							)]
        a__	:					Dict										=feature_extractor(
            lowerCAmelCase__		,      padding="max_length"		,      max_length=4		,      truncation=lowerCAmelCase__		,      return_tensors="np"		,      return_attention_mask=lowerCAmelCase__		,      )
        a__	:					Optional[int]										=inputs.input_features
        a__	:					Tuple										=inputs.attention_mask
        a__	:					Optional[Any]										=np.sum(attention_mask == 1		,      axis=1							)

        self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]							)
        self._check_zero_mean_unit_variance(input_features[1]							)
        self._check_zero_mean_unit_variance(input_features[2]							)







      def    _lowercase  (    self							)   ->						Dict:



        '''simple docstring'''




        a__	:					Any										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        a__	:					Tuple										=[floats_list((1, x)							)[0] for x in range(8_0_0		,      1_4_0_0		,      2_0_0							)]
        a__	:					Dict										=feature_extractor(
            lowerCAmelCase__		,      padding="longest"		,      max_length=4		,      truncation=lowerCAmelCase__		,      return_tensors="np"		,      return_attention_mask=lowerCAmelCase__		,      )
        a__	:					Union[str, Any]										=inputs.input_features
        a__	:					Union[str, Any]										=inputs.attention_mask
        a__	:					List[Any]										=np.sum(attention_mask == 1		,      axis=1							)

        self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]							)
        self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]							)
        self._check_zero_mean_unit_variance(input_features[2]							)

        # make sure that if max_length < longest -> then pad to max_length
        self.assertEqual(input_features.shape		,      (3, 4, 2_4)							)

        a__	:					List[Any]										=[floats_list((1, x)							)[0] for x in range(8_0_0		,      1_4_0_0		,      2_0_0							)]
        a__	:					Any										=feature_extractor(
            lowerCAmelCase__		,      padding="longest"		,      max_length=1_6		,      truncation=lowerCAmelCase__		,      return_tensors="np"		,      return_attention_mask=lowerCAmelCase__		,      )
        a__	:					Tuple										=inputs.input_features
        a__	:					List[Any]										=inputs.attention_mask
        a__	:					int										=np.sum(attention_mask == 1		,      axis=1							)

        self._check_zero_mean_unit_variance(input_features[0, : fbank_feat_lengths[0]]							)
        self._check_zero_mean_unit_variance(input_features[1, : fbank_feat_lengths[1]]							)
        self._check_zero_mean_unit_variance(input_features[2]							)

        # make sure that if max_length < longest -> then pad to max_length
        self.assertEqual(input_features.shape		,      (3, 6, 2_4)							)







      def    _lowercase  (    self							)   ->						int:



        '''simple docstring'''




        import torch

        a__	:					Union[str, Any]										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        a__	:					Dict										=np.random.rand(1_0_0		,      3_2							).astype(np.floataa							)
        a__	:					List[str]										=np_speech_inputs.tolist()

        for inputs in [py_speech_inputs, np_speech_inputs]:
          a__	:					List[str]										=feature_extractor.pad([{"input_features": inputs}]		,      return_tensors="np"							)
          self.assertTrue(np_processed.input_features.dtype == np.floataa							)
          a__	:					Tuple										=feature_extractor.pad([{"input_features": inputs}]		,      return_tensors="pt"							)
          self.assertTrue(pt_processed.input_features.dtype == torch.floataa							)







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



        '''simple docstring'''




        from datasets import load_dataset

        a__	:					Tuple										=load_dataset("hf-internal-testing/librispeech_asr_dummy"		,      "clean"		,      split="validation"							)
        # automatic decoding with librispeech
        a__	:					List[str]										=ds.sort("id"							).select(range(lowerCAmelCase__							)							)[:num_samples]["audio"]

        return [x["array"] for x in speech_samples]







      def    _lowercase  (    self							)   ->						List[str]:



        '''simple docstring'''




        a__	:					Any										=np.array([
            -1.57_45, -1.77_13, -1.70_20, -1.60_69, -1.22_50, -1.11_05, -0.90_72, -0.82_41,
            -1.23_10, -0.80_98, -0.33_20, -0.41_01, -0.79_85, -0.49_96, -0.82_13, -0.91_28,
            -1.04_20, -1.12_86, -1.04_40, -0.79_99, -0.84_05, -1.22_75, -1.54_43, -1.46_25,
        ]							)
        # fmt: on

        a__	:					Any										=self._load_datasamples(1							)
        a__	:					Any										=self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()							)
        a__	:					Tuple										=feature_extractor(lowerCAmelCase__		,      return_tensors="pt"							).input_features
        self.assertEquals(input_features.shape		,      (1, 5_8_4, 2_4)							)
        self.assertTrue(np.allclose(input_features[0, 0, :3_0]		,      lowerCAmelCase__		,      atol=1E-4							)							)













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






