

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


A_		:      Optional[int]					    = random.Random()


def    __a							( SCREAMING_SNAKE_CASE							, SCREAMING_SNAKE_CASE=1.0							, SCREAMING_SNAKE_CASE=None							, SCREAMING_SNAKE_CASE=None   )   ->  Dict:






    '''simple docstring'''
    if rng is None:
        __UpperCAmelCase					       =      global_rng

    __UpperCAmelCase					       =      []
    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  ,   lowercase__  ,   lowercase__=7  ,   lowercase__=400  ,   lowercase__=2_000  ,   lowercase__=1  ,   lowercase__=0.0  ,   lowercase__=16_000  ,   lowercase__=True  ,   lowercase__=True  ,   )  ->   str:
        __UpperCAmelCase					       =      parent
        __UpperCAmelCase					       =      batch_size
        __UpperCAmelCase					       =      min_seq_length
        __UpperCAmelCase					       =      max_seq_length
        __UpperCAmelCase					       =      (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        __UpperCAmelCase					       =      feature_size
        __UpperCAmelCase					       =      padding_value
        __UpperCAmelCase					       =      sampling_rate
        __UpperCAmelCase					       =      return_attention_mask
        __UpperCAmelCase					       =      do_normalize


    def   lowerCAmelCase_    (self	)  ->   str:
        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   lowerCAmelCase_    (self  ,   lowercase__=False  ,   lowercase__=False	)  ->   str:
        def _flatten(lowercase__	):
            return list(itertools.chain(*A_	)	)

        if equal_length:
            __UpperCAmelCase					       =      floats_list((self.batch_size, self.max_seq_length)	)
        else:
            # make sure that inputs increase in size
            __UpperCAmelCase					       =      [
                _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:
            __UpperCAmelCase					       =      [np.asarray(A_	) for x in speech_inputs]

        return speech_inputs






class        A_ (   _a			,  unittest.TestCase    ):


    '''simple docstring'''






    a__  				=							WavaVecaFeatureExtractor


    def   lowerCAmelCase_    (self	)  ->   List[Any]:
        __UpperCAmelCase					       =      WavaVecaFeatureExtractionTester(self	)


    def   lowerCAmelCase_    (self  ,   lowercase__	)  ->   int:
        self.assertTrue(np.all(np.mean(A_  ,   axis=0	) < 1E-3	)	)
        self.assertTrue(np.all(np.abs(np.var(A_  ,   axis=0	) - 1	) < 1E-3	)	)


    def   lowerCAmelCase_    (self	)  ->   List[str]:
        # Tests that all call wrap to encode_plus and batch_encode_plus
        __UpperCAmelCase					       =      self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
        # create three inputs of length 800, 1000, and 1200
        __UpperCAmelCase					       =      [floats_list((1, x)	)[0] for x in range(800  ,   1_400  ,   200	)]
        __UpperCAmelCase					       =      [np.asarray(A_	) for speech_input in speech_inputs]

        # Test not batched input
        __UpperCAmelCase					       =      feat_extract(speech_inputs[0]  ,   return_tensors='''np'''	).input_values
        __UpperCAmelCase					       =      feat_extract(np_speech_inputs[0]  ,   return_tensors='''np'''	).input_values
        self.assertTrue(np.allclose(A_  ,   A_  ,   atol=1E-3	)	)

        # Test batched
        __UpperCAmelCase					       =      feat_extract(A_  ,   return_tensors='''np'''	).input_values
        __UpperCAmelCase					       =      feat_extract(A_  ,   return_tensors='''np'''	).input_values
        for enc_seq_a, enc_seq_a in zip(A_  ,   A_	):
            self.assertTrue(np.allclose(A_  ,   A_  ,   atol=1E-3	)	)

        # Test 2-D numpy arrays are batched.
        __UpperCAmelCase					       =      [floats_list((1, x)	)[0] for x in (800, 800, 800)]
        __UpperCAmelCase					       =      np.asarray(A_	)
        __UpperCAmelCase					       =      feat_extract(A_  ,   return_tensors='''np'''	).input_values
        __UpperCAmelCase					       =      feat_extract(A_  ,   return_tensors='''np'''	).input_values
        for enc_seq_a, enc_seq_a in zip(A_  ,   A_	):
            self.assertTrue(np.allclose(A_  ,   A_  ,   atol=1E-3	)	)


    def   lowerCAmelCase_    (self	)  ->   str:
        __UpperCAmelCase					       =      self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
        __UpperCAmelCase					       =      [floats_list((1, x)	)[0] for x in range(800  ,   1_400  ,   200	)]

        __UpperCAmelCase					       =      ['''longest''', '''max_length''', '''do_not_pad''']
        __UpperCAmelCase					       =      [None, 1_600, None]
        for max_length, padding in zip(A_  ,   A_	):
            __UpperCAmelCase					       =      feat_extract(A_  ,   padding=A_  ,   max_length=A_  ,   return_tensors='''np'''	)
            __UpperCAmelCase					       =      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   lowerCAmelCase_    (self	)  ->   List[Any]:
        __UpperCAmelCase					       =      self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
        __UpperCAmelCase					       =      range(800  ,   1_400  ,   200	)
        __UpperCAmelCase					       =      [floats_list((1, x)	)[0] for x in lengths]

        __UpperCAmelCase					       =      ['''longest''', '''max_length''', '''do_not_pad''']
        __UpperCAmelCase					       =      [None, 1_600, None]

        for max_length, padding in zip(A_  ,   A_	):
            __UpperCAmelCase					       =      feat_extract(A_  ,   max_length=A_  ,   padding=A_	)
            __UpperCAmelCase					       =      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   lowerCAmelCase_    (self	)  ->   str:
        __UpperCAmelCase					       =      self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
        __UpperCAmelCase					       =      [floats_list((1, x)	)[0] for x in range(800  ,   1_400  ,   200	)]
        __UpperCAmelCase					       =      feat_extract(
            A_  ,   truncation=A_  ,   max_length=1_000  ,   padding='''max_length'''  ,   return_tensors='''np'''	)
        __UpperCAmelCase					       =      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   lowerCAmelCase_    (self	)  ->   int:
        __UpperCAmelCase					       =      self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()	)
        __UpperCAmelCase					       =      [floats_list((1, x)	)[0] for x in range(800  ,   1_400  ,   200	)]
        __UpperCAmelCase					       =      feat_extract(
            A_  ,   truncation=A_  ,   max_length=1_000  ,   padding='''longest'''  ,   return_tensors='''np'''	)
        __UpperCAmelCase					       =      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)	)

        __UpperCAmelCase					       =      [floats_list((1, x)	)[0] for x in range(800  ,   1_400  ,   200	)]
        __UpperCAmelCase					       =      feat_extract(
            A_  ,   truncation=A_  ,   max_length=2_000  ,   padding='''longest'''  ,   return_tensors='''np'''	)
        __UpperCAmelCase					       =      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   lowerCAmelCase_    (self	)  ->   Union[str, Any]:
        import torch

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

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






    @slow
    @require_torch
    def   lowerCAmelCase_    (self	)  ->   Optional[int]:
        # this test makes sure that models that are using
        # group norm don't have their feature extractor return the
        # attention_mask
        for model_id in WAV_2_VEC_2_PRETRAINED_MODEL_ARCHIVE_LIST:
            __UpperCAmelCase					       =      WavaVecaConfig.from_pretrained(A_	)
            __UpperCAmelCase					       =      WavaVecaFeatureExtractor.from_pretrained(A_	)

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








"""simple docstring"""




import json
import os
from pathlib import Path

import pytest

from datasets.download.download_config import DownloadConfig
from datasets.download.download_manager import DownloadManager
from datasets.utils.file_utils import hash_url_to_filename


__magic_name__						: Optional[Any]    =				'http://www.mocksite.com/file1.txt'
__magic_name__						: Tuple    =				'"text": ["foo", "foo"]'
__magic_name__						: str    =				'6d8ce9aa78a471c7477201efbeabd3bb01ac2e7d100a6dc024ba1608361f90a8'




class 			__snake_case   :
							__a					 =       200
							__a					 =       {'''Content-Length''': '''100'''}
							__a					 =       {}




							def   __a   (     self:    List[str]     ,							**A_:    List[Any]   ):
										return [bytes(A_     ,							"""utf-8"""   )]





def       a_ (	*lowercase__ :List[Any],			**lowercase__ :Any     ):
			return MockResponse()





@pytest.mark.parametrize("""urls_type""",			[str, list, dict]     )
def       a_ (	lowercase__ :Optional[int],			lowercase__ :Any,			lowercase__ :Optional[int]     ):
			import requests

			monkeypatch.setattr(lowercase__,			"""request""",			lowercase__     )

			__lowerCamelCase          =      URL
			if issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      url
			elif issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      [url]
			elif issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      {"""train""": url}
			__lowerCamelCase          =      """dummy"""
			__lowerCamelCase          =      """downloads"""
			__lowerCamelCase          =      tmp_path
			__lowerCamelCase          =      DownloadConfig(
			    cache_dir=os.path.join(lowercase__,			lowercase__     ),			use_etag=lowercase__,			)
			__lowerCamelCase          =      DownloadManager(dataset_name=lowercase__,			download_config=lowercase__     )
			__lowerCamelCase          =      dl_manager.download(lowercase__     )
			__lowerCamelCase          =      urls
			for downloaded_paths in [downloaded_paths]:
						if isinstance(lowercase__,			lowercase__     ):
									__lowerCamelCase          =      [downloaded_paths]
									__lowerCamelCase          =      [urls]
						elif isinstance(lowercase__,			lowercase__     ):
									assert "train" in downloaded_paths.keys()
									__lowerCamelCase          =      downloaded_paths.values()
									__lowerCamelCase          =      urls.values()
						assert downloaded_paths
						for downloaded_path, input_url in zip(lowercase__,			lowercase__     ):
									assert downloaded_path == dl_manager.downloaded_paths[input_url]
									__lowerCamelCase          =      Path(lowercase__     )
									__lowerCamelCase          =      downloaded_path.parts
									assert parts[-1] == HASH
									assert parts[-2] == cache_subdir
									assert downloaded_path.exists()
									__lowerCamelCase          =      downloaded_path.read_text()
									assert content == CONTENT
									__lowerCamelCase          =      downloaded_path.with_suffix(""".json"""     )
									assert metadata_downloaded_path.exists()
									__lowerCamelCase          =      json.loads(metadata_downloaded_path.read_text()     )
									assert metadata_content == {"url": URL, "etag": None}





@pytest.mark.parametrize("""paths_type""",			[str, list, dict]     )
def       a_ (	lowercase__ :Dict,			lowercase__ :Optional[Any],			lowercase__ :Dict     ):
			__lowerCamelCase          =      str(lowercase__     )
			if issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      filename
			elif issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      [filename]
			elif issubclass(lowercase__,			lowercase__     ):
						__lowerCamelCase          =      {"""train""": filename}
			__lowerCamelCase          =      """dummy"""
			__lowerCamelCase          =      xz_file.parent
			__lowerCamelCase          =      """extracted"""
			__lowerCamelCase          =      DownloadConfig(
			    cache_dir=lowercase__,			use_etag=lowercase__,			)
			__lowerCamelCase          =      DownloadManager(dataset_name=lowercase__,			download_config=lowercase__     )
			__lowerCamelCase          =      dl_manager.extract(lowercase__     )
			__lowerCamelCase          =      paths
			for extracted_paths in [extracted_paths]:
						if isinstance(lowercase__,			lowercase__     ):
									__lowerCamelCase          =      [extracted_paths]
									__lowerCamelCase          =      [paths]
						elif isinstance(lowercase__,			lowercase__     ):
									assert "train" in extracted_paths.keys()
									__lowerCamelCase          =      extracted_paths.values()
									__lowerCamelCase          =      paths.values()
						assert extracted_paths
						for extracted_path, input_path in zip(lowercase__,			lowercase__     ):
									assert extracted_path == dl_manager.extracted_paths[input_path]
									__lowerCamelCase          =      Path(lowercase__     )
									__lowerCamelCase          =      extracted_path.parts
									assert parts[-1] == hash_url_to_filename(lowercase__,			etag=lowercase__     )
									assert parts[-2] == extracted_subdir
									assert extracted_path.exists()
									__lowerCamelCase          =      extracted_path.read_text()
									__lowerCamelCase          =      text_file.read_text()
									assert extracted_file_content == expected_file_content





def       a_ (	lowercase__ :List[str],			lowercase__ :int     ):
			assert path.endswith(""".jsonl"""     )
			for num_items, line in enumerate(lowercase__,			start=1     ):
						__lowerCamelCase          =      json.loads(line.decode("""utf-8"""     )     )
						assert item.keys() == {"col_1", "col_2", "col_3"}
			assert num_items == 4





@pytest.mark.parametrize("""archive_jsonl""",			["""tar_jsonl_path""", """zip_jsonl_path"""]     )
def       a_ (	lowercase__ :Optional[int],			lowercase__ :Union[str, Any]     ):
			__lowerCamelCase          =      request.getfixturevalue(lowercase__     )
			__lowerCamelCase          =      DownloadManager()
			for num_jsonl, (path, file) in enumerate(dl_manager.iter_archive(lowercase__     ),			start=1     ):
						_test_jsonl(lowercase__,			lowercase__     )
			assert num_jsonl == 2





@pytest.mark.parametrize("""archive_nested_jsonl""",			["""tar_nested_jsonl_path""", """zip_nested_jsonl_path"""]     )
def       a_ (	lowercase__ :Optional[int],			lowercase__ :List[Any]     ):
			__lowerCamelCase          =      request.getfixturevalue(lowercase__     )
			__lowerCamelCase          =      DownloadManager()
			for num_tar, (path, file) in enumerate(dl_manager.iter_archive(lowercase__     ),			start=1     ):
						for num_jsonl, (subpath, subfile) in enumerate(dl_manager.iter_archive(lowercase__     ),			start=1     ):
									_test_jsonl(lowercase__,			lowercase__     )
			assert num_tar == 1
			assert num_jsonl == 2





def       a_ (	lowercase__ :Tuple     ):
			__lowerCamelCase          =      DownloadManager()
			for num_file, file in enumerate(dl_manager.iter_files(lowercase__     ),			start=1     ):
						assert os.path.basename(lowercase__     ) == ("test.txt" if num_file == 1 else "train.txt")
			assert num_file == 2




