



import itertools
import random
import unittest

import numpy as np

from transformers import WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST, WavaVecaConfig, WavaVecaFeatureExtractor
from transformers.testing_utils import require_torch, slow

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


snake_case__					:     Tuple    		=   random.Random()



def    snake_case_				(    _SCREAMING_SNAKE_CASE  ,	_SCREAMING_SNAKE_CASE=1.0  ,	_SCREAMING_SNAKE_CASE=None  ,	_SCREAMING_SNAKE_CASE=None   ):





 if rng is None:
  __lowercase		       =			global_rng

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

 return values

class   _A   ( unittest.TestCase	):


      '''simple docstring'''





      def __init__(	self	:							List[Any]							,      lowerCamelCase	:							Union[str, Any]							,      lowerCamelCase	:							Tuple=7							,      lowerCamelCase	:							Union[str, Any]=400							,      lowerCamelCase	:							Optional[int]=2_000							,      lowerCamelCase	:							Union[str, Any]=1							,      lowerCamelCase	:							Tuple=0.0							,      lowerCamelCase	:							Union[str, Any]=16_000							,      lowerCamelCase	:							Any=True							,      lowerCamelCase	:							str=True							,      ):






       '''simple docstring'''



       __lowercase		       =			parent
       __lowercase		       =			batch_size
       __lowercase		       =			min_seq_length
       __lowercase		       =			max_seq_length
       __lowercase		       =			(self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
       __lowercase		       =			feature_size
       __lowercase		       =			padding_value
       __lowercase		       =			sampling_rate
       __lowercase		       =			return_attention_mask
       __lowercase		       =			do_normalize



      def     _snake_case	(	self	:							List[Any]	):






       '''simple docstring'''



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



      def     _snake_case	(	self	:							Optional[int]							,      lowerCamelCase	:							List[str]=False							,      lowerCamelCase	:							Union[str, Any]=False	):






       '''simple docstring'''



       def _flatten(lowerCamelCase	:							Optional[Any]	):
        return list(itertools.chain(*lowerCamelCase	)	)

       if equal_length:
        __lowercase		       =			floats_list((self.batch_size, self.max_seq_length)	)
       else:
        # make sure that inputs increase in size
        __lowercase		       =			[
            _flatten(floats_list((x, self.feature_size)	)	)
            for x in range(self.min_seq_length							,      self.max_seq_length							,      self.seq_length_diff	)
        ]

       if numpify:
        __lowercase		       =			[np.asarray(lowerCamelCase	) for x in speech_inputs]

       return speech_inputs

class   _A   ( __lowercase  ,						unittest.TestCase	):


      '''simple docstring'''





      _snake_case			:     List[Any]      							=      WavaVecaFeatureExtractor



      def     _snake_case	(	self	:							Optional[Any]	):






       '''simple docstring'''



       __lowercase		       =			WavaVecaFeatureExtractionTester(self	)



      def     _snake_case	(	self	:							Union[str, Any]							,      lowerCamelCase	:							Union[str, 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     _snake_case	(	self	:							Union[str, Any]	):






       '''simple docstring'''



       __lowercase		       =			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
       # create three inputs of length 800, 1000, and 1200
       __lowercase		       =			[floats_list((1, x)	)[0] for x in range(800							,      1_400							,      200	)]
       __lowercase		       =			[np.asarray(lowerCamelCase	) for speech_input in speech_inputs]

       # Test not batched input
       __lowercase		       =			feat_extract(speech_inputs[0]							,      return_tensors="np"	).input_values
       __lowercase		       =			feat_extract(np_speech_inputs[0]							,      return_tensors="np"	).input_values
       self.assertTrue(np.allclose(lowerCamelCase							,      lowerCamelCase							,      atol=1e-3	)	)

       # Test batched
       __lowercase		       =			feat_extract(lowerCamelCase							,      return_tensors="np"	).input_values
       __lowercase		       =			feat_extract(lowerCamelCase							,      return_tensors="np"	).input_values
       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.
       __lowercase		       =			[floats_list((1, x)	)[0] for x in (800, 800, 800)]
       __lowercase		       =			np.asarray(lowerCamelCase	)
       __lowercase		       =			feat_extract(lowerCamelCase							,      return_tensors="np"	).input_values
       __lowercase		       =			feat_extract(lowerCamelCase							,      return_tensors="np"	).input_values
       for enc_seq_a, enc_seq_a in zip(lowerCamelCase							,      lowerCamelCase	):
        self.assertTrue(np.allclose(lowerCamelCase							,      lowerCamelCase							,      atol=1e-3	)	)



      def     _snake_case	(	self	:							Optional[Any]	):






       '''simple docstring'''



       __lowercase		       =			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
       __lowercase		       =			[floats_list((1, x)	)[0] for x in range(800							,      1_400							,      200	)]

       __lowercase		       =			["longest", "max_length", "do_not_pad"]
       __lowercase		       =			[None, 1_600, None]
       for max_length, padding in zip(lowerCamelCase							,      lowerCamelCase	):
        __lowercase		       =			feat_extract(lowerCamelCase							,      padding=lowerCamelCase							,      max_length=lowerCamelCase							,      return_tensors="np"	)
        __lowercase		       =			processed.input_values

        self._check_zero_mean_unit_variance(input_values[0][:800]	)
        self.assertTrue(input_values[0][800:].sum() < 1e-6	)
        self._check_zero_mean_unit_variance(input_values[1][:1_000]	)
        self.assertTrue(input_values[0][1_000:].sum() < 1e-6	)
        self._check_zero_mean_unit_variance(input_values[2][:1_200]	)



      def     _snake_case	(	self	:							Dict	):






       '''simple docstring'''



       __lowercase		       =			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
       __lowercase		       =			range(800							,      1_400							,      200	)
       __lowercase		       =			[floats_list((1, x)	)[0] for x in lengths]

       __lowercase		       =			["longest", "max_length", "do_not_pad"]
       __lowercase		       =			[None, 1_600, None]

       for max_length, padding in zip(lowerCamelCase							,      lowerCamelCase	):
        __lowercase		       =			feat_extract(lowerCamelCase							,      max_length=lowerCamelCase							,      padding=lowerCamelCase	)
        __lowercase		       =			processed.input_values

        self._check_zero_mean_unit_variance(input_values[0][:800]	)
        self._check_zero_mean_unit_variance(input_values[1][:1_000]	)
        self._check_zero_mean_unit_variance(input_values[2][:1_200]	)



      def     _snake_case	(	self	:							List[str]	):






       '''simple docstring'''



       __lowercase		       =			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
       __lowercase		       =			[floats_list((1, x)	)[0] for x in range(800							,      1_400							,      200	)]
       __lowercase		       =			feat_extract(
           lowerCamelCase							,      truncation=lowerCamelCase							,      max_length=1_000							,      padding="max_length"							,      return_tensors="np"	)
       __lowercase		       =			processed.input_values

       self._check_zero_mean_unit_variance(input_values[0, :800]	)
       self._check_zero_mean_unit_variance(input_values[1]	)
       self._check_zero_mean_unit_variance(input_values[2]	)



      def     _snake_case	(	self	:							int	):






       '''simple docstring'''



       __lowercase		       =			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
       __lowercase		       =			[floats_list((1, x)	)[0] for x in range(800							,      1_400							,      200	)]
       __lowercase		       =			feat_extract(
           lowerCamelCase							,      truncation=lowerCamelCase							,      max_length=1_000							,      padding="longest"							,      return_tensors="np"	)
       __lowercase		       =			processed.input_values

       self._check_zero_mean_unit_variance(input_values[0, :800]	)
       self._check_zero_mean_unit_variance(input_values[1, :1_000]	)
       self._check_zero_mean_unit_variance(input_values[2]	)

       # make sure that if max_length < longest -> then pad to max_length
       self.assertTrue(input_values.shape == (3, 1_000)	)

       __lowercase		       =			[floats_list((1, x)	)[0] for x in range(800							,      1_400							,      200	)]
       __lowercase		       =			feat_extract(
           lowerCamelCase							,      truncation=lowerCamelCase							,      max_length=2_000							,      padding="longest"							,      return_tensors="np"	)
       __lowercase		       =			processed.input_values

       self._check_zero_mean_unit_variance(input_values[0, :800]	)
       self._check_zero_mean_unit_variance(input_values[1, :1_000]	)
       self._check_zero_mean_unit_variance(input_values[2]	)

       # make sure that if max_length > longest -> then pad to longest
       self.assertTrue(input_values.shape == (3, 1_200)	)



      @require_torch
      def     _snake_case	(	self	:							Optional[int]	):






       '''simple docstring'''



       import torch

       __lowercase		       =			self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
       __lowercase		       =			np.random.rand(100	).astype(np.floataa	)
       __lowercase		       =			np_speech_inputs.tolist()

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



      @slow
      @require_torch
      def     _snake_case	(	self	:							Any	):






       '''simple docstring'''



       for model_id in WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST:
        __lowercase		       =			WavaVecaConfig.from_pretrained(lowerCamelCase	)
        __lowercase		       =			WavaVecaFeatureExtractor.from_pretrained(lowerCamelCase	)

        # only "layer" feature extraction norm should make use of
        # attention_mask
        self.assertEqual(feat_extract.return_attention_mask							,      config.feat_extract_norm == "layer"	)








from __future__ import annotations

from collections.abc import Callable
from typing import Generic, TypeVar

snake_case__					:     Union[str, Any]    		=   TypeVar("""T""")
snake_case__					:     Optional[int]    		=   TypeVar("""U""")

class   _A   ( Generic[T, U]	):


						'''simple docstring'''





						def __init__(	self	:							Optional[int]							,      lowerCamelCase	:							T | None							,      lowerCamelCase	:							U | None	):






							'''simple docstring'''



							__lowercase		       =			key
							__lowercase		       =			val
							__lowercase		       =			None
							__lowercase		       =			None



						def __repr__(	self	:							Any	):






							'''simple docstring'''



							return (
							    f"""Node: key: {self.key}, val: {self.val}, """
							    f"""has next: {bool(self.next	)}, has prev: {bool(self.prev	)}"""
							)

class   _A   ( Generic[T, U]	):


						'''simple docstring'''





						def __init__(	self	:							Dict	):






							'''simple docstring'''



							__lowercase		       =			DoubleLinkedListNode(lowerCamelCase							,      lowerCamelCase	)
							__lowercase		       =			DoubleLinkedListNode(lowerCamelCase							,      lowerCamelCase	)
							__lowercase       ,    __lowercase		       =			self.rear, self.head



						def __repr__(	self	:							Optional[Any]	):






							'''simple docstring'''



							__lowercase		       =			["DoubleLinkedList"]
							__lowercase		       =			self.head
							while node.next is not None:
								rep.append(str(lowerCamelCase	)	)
								__lowercase		       =			node.next
							rep.append(str(self.rear	)	)
							return ",\n    ".join(lowerCamelCase	)



						def     _snake_case	(	self	:							Union[str, Any]							,      lowerCamelCase	:							DoubleLinkedListNode[T, U]	):






							'''simple docstring'''



							__lowercase		       =			self.rear.prev

							# All nodes other than self.head are guaranteed to have non-None previous
							assert previous is not None

							__lowercase		       =			node
							__lowercase		       =			previous
							__lowercase		       =			node
							__lowercase		       =			self.rear



						def     _snake_case	(	self	:							Optional[int]							,      lowerCamelCase	:							DoubleLinkedListNode[T, U]	):






							'''simple docstring'''



							if node.prev is None or node.next is None:
								return None

							__lowercase		       =			node.next
							__lowercase		       =			node.prev
							__lowercase		       =			None
							__lowercase		       =			None
							return node

class   _A   ( Generic[T, U]	):


						'''simple docstring'''





						_snake_case			:     dict[Callable[[T], U], LRUCache[T, U]]      							=      {}



						def __init__(	self	:							List[Any]							,      lowerCamelCase	:							int	):






							'''simple docstring'''



							__lowercase		       =			DoubleLinkedList()
							__lowercase		       =			capacity
							__lowercase		       =			0
							__lowercase		       =			0
							__lowercase		       =			0
							__lowercase		       =			{}



						def __repr__(	self	:							Optional[Any]	):






							'''simple docstring'''



							return (
							    f"""CacheInfo(hits={self.hits}, misses={self.miss}, """
							    f"""capacity={self.capacity}, current size={self.num_keys})"""
							)



						def __contains__(	self	:							Dict							,      lowerCamelCase	:							T	):






							'''simple docstring'''



							return key in self.cache



						def     _snake_case	(	self	:							List[Any]							,      lowerCamelCase	:							T	):






							'''simple docstring'''



							if key in self.cache:
								self.hits += 1
								__lowercase		       =			self.cache[key]
								__lowercase		       =			self.list.remove(self.cache[key]	)
								assert node == value_node

								# node is guaranteed not None because it is in self.cache
								assert node is not None
								self.list.add(lowerCamelCase	)
								return node.val
							self.miss += 1
							return None



						def     _snake_case	(	self	:							Union[str, Any]							,      lowerCamelCase	:							T							,      lowerCamelCase	:							U	):






							'''simple docstring'''



							if key not in self.cache:
								if self.num_keys >= self.capacity:
									# delete first node (oldest) when over capacity
									__lowercase		       =			self.list.head.next

									# guaranteed to have a non-None first node when num_keys > 0
									# explain to type checker via assertions
									assert first_node is not None
									assert first_node.key is not None
									assert (
									    self.list.remove(lowerCamelCase	) is not None
									)  # node guaranteed to be in list assert node.key is not None

									del self.cache[first_node.key]
									self.num_keys -= 1
								__lowercase		       =			DoubleLinkedListNode(lowerCamelCase							,      lowerCamelCase	)
								self.list.add(self.cache[key]	)
								self.num_keys += 1

							else:
								# bump node to the end of the list, update value
								__lowercase		       =			self.list.remove(self.cache[key]	)
								assert node is not None  # node guaranteed to be in list
								__lowercase		       =			value
								self.list.add(lowerCamelCase	)



						@classmethod
						def     _snake_case	(	cls	:							Union[str, Any]							,      lowerCamelCase	:							int = 128	):






							'''simple docstring'''



							def cache_decorator_inner(lowerCamelCase	:							Callable[[T], U]	) -> Callable[..., U]:
								def cache_decorator_wrapper(*lowerCamelCase	:							T	) -> U:
									if func not in cls.decorator_function_to_instance_map:
										__lowercase		       =			LRUCache(lowerCamelCase	)

									__lowercase		       =			cls.decorator_function_to_instance_map[func].get(args[0]	)
									if result is None:
										__lowercase		       =			func(*lowerCamelCase	)
										cls.decorator_function_to_instance_map[func].put(args[0]							,      lowerCamelCase	)
									return result

								def cache_info() -> LRUCache[T, U]:
									return cls.decorator_function_to_instance_map[func]

								setattr(lowerCamelCase							,      "cache_info"							,      lowerCamelCase	)  # noqa: B010

								return cache_decorator_wrapper

							return cache_decorator_inner


if __name__ == "__main__":
	import doctest

	doctest.testmod()



