import itertools
import random
import unittest

import numpy as np

from transformers import EncodecFeatureExtractor
from transformers.testing_utils import require_torch
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
  import torch


UpperCAmelCase				 =   random.Random()







def 						UpperCAmelCase							(		lowerCAmelCase_    ,   lowerCAmelCase_=1.0    ,   lowerCAmelCase_=None    ,   lowerCAmelCase_=None							)						->						Any:
     if rng is None:
          a	:     str						      =    global_rng

     a	:     Union[str, 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
class 				_SCREAMING_SNAKE_CASE( unittest.TestCase						):
   def __init__(     self				,  lowerCAmelCase_				,  lowerCAmelCase_=7				,  lowerCAmelCase_=4_00				,  lowerCAmelCase_=20_00				,  lowerCAmelCase_=1				,  lowerCAmelCase_=0.0				,  lowerCAmelCase_=2_40_00				,  lowerCAmelCase_=True				,  )						->Optional[Any]:
        a	:     Optional[Any]						      =    parent
        a	:     List[str]						      =    batch_size
        a	:     int						      =    min_seq_length
        a	:     str						      =    max_seq_length
        a	:     Any						      =    (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
        a	:     Optional[Any]						      =    feature_size
        a	:     Optional[Any]						      =    padding_value
        a	:     Union[str, Any]						      =    sampling_rate
        a	:     Optional[int]						      =    return_attention_mask
   def _lowercase(     self     )						->Optional[int]:
        return {
            "feature_size": self.feature_size,
            "padding_value": self.padding_value,
            "sampling_rate": self.sampling_rate,
            "return_attention_mask": self.return_attention_mask,
        }


   def _lowercase(     self				,  lowerCAmelCase_=False				,  lowerCAmelCase_=False     )						->Union[str, Any]:
        def _flatten(lowerCAmelCase_     ):
             return list(itertools.chain(*_lowercase     )     )

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

        return audio_inputs






@require_torch
class 				_SCREAMING_SNAKE_CASE( __lowerCamelCase							,   unittest.TestCase						):
   UpperCamelCase_					:    Optional[int]			      = EncodecFeatureExtractor
   def _lowercase(     self     )						->List[str]:
        a	:     Dict						      =    EnCodecFeatureExtractionTester(self     )
   def _lowercase(     self     )						->Any:
        a	:     Any						      =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()     )
        # create three inputs of length 800, 1000, and 1200
        a	:     Tuple						      =    [floats_list((1, x)     )[0] for x in range(8_00				,  14_00				,  2_00     )]
        a	:     Union[str, Any]						      =    [np.asarray(_lowercase     ) for audio_input in audio_inputs]

        # Test not batched input
        a	:     Optional[int]						      =    feat_extract(audio_inputs[0]				,  return_tensors='''np'''     ).input_values
        a	:     int						      =    feat_extract(np_audio_inputs[0]				,  return_tensors='''np'''     ).input_values
        self.assertTrue(np.allclose(_lowercase				,  _lowercase				,  atol=1e-3     )     )

        # Test batched
        a	:     Optional[Any]						      =    feat_extract(_lowercase				,  padding=_lowercase				,  return_tensors='''np'''     ).input_values
        a	:     List[Any]						      =    feat_extract(_lowercase				,  padding=_lowercase				,  return_tensors='''np'''     ).input_values
        for enc_seq_a, enc_seq_a in zip(_lowercase				,  _lowercase     ):
             self.assertTrue(np.allclose(_lowercase				,  _lowercase				,  atol=1e-3     )     )
   def _lowercase(     self     )						->Union[str, Any]:
        a	:     int						      =    self.feature_extraction_class(**self.feat_extract_tester.prepare_feat_extract_dict()     )
        a	:     str						      =    np.random.rand(1_00     ).astype(np.floataa     )
        a	:     int						      =    np_audio_inputs.tolist()

        for inputs in [py_audio_inputs, np_audio_inputs]:
             a	:     Optional[Any]						      =    feature_extractor.pad([{'''input_values''': inputs}]				,  return_tensors='''np'''     )
             self.assertTrue(np_processed.input_values.dtype == np.floataa     )
             a	:     Optional[int]						      =    feature_extractor.pad([{'''input_values''': inputs}]				,  return_tensors='''pt'''     )
             self.assertTrue(pt_processed.input_values.dtype == torch.floataa     )
   def _lowercase(     self				,  lowerCAmelCase_     )						->List[str]:
        from datasets import load_dataset

        a	:     List[str]						      =    load_dataset('''hf-internal-testing/librispeech_asr_dummy'''				,  '''clean'''				,  split='''validation'''     )
        # automatic decoding with librispeech
        a	:     Optional[int]						      =    ds.sort('''id'''     ).select(range(_lowercase     )     )[:num_samples]['''audio''']

        return [x["array"] for x in audio_samples]
   def _lowercase(     self     )						->Union[str, Any]:
        a	:     List[Any]						      =    torch.tensor(
            [2.3_804e-03, 2.0_752e-03, 1.9_836e-03, 2.1_057e-03, 1.6_174e-03,
             3.0_518e-04, 9.1_553e-05, 3.3_569e-04, 9.7_656e-04, 1.8_311e-03,
             2.0_142e-03, 2.1_057e-03, 1.7_395e-03, 4.5_776e-04, -3.9_673e-04,
             4.5_776e-04, 1.0_071e-03, 9.1_553e-05, 4.8_828e-04, 1.1_597e-03,
             7.3_242e-04, 9.4_604e-04, 1.8_005e-03, 1.8_311e-03, 8.8_501e-04,
             4.2_725e-04, 4.8_828e-04, 7.3_242e-04, 1.0_986e-03, 2.1_057e-03]     )
        # fmt: on
        a	:     Tuple						      =    self._load_datasamples(1     )
        a	:     Union[str, Any]						      =    EncodecFeatureExtractor()
        a	:     Optional[Any]						      =    feature_extractor(_lowercase				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(input_values.shape				,  (1, 1, 9_36_80)     )
        self.assertTrue(torch.allclose(input_values[0, 0, :30]				,  _lowercase				,  atol=1e-6     )     )
   def _lowercase(     self     )						->Any:
        a	:     Optional[int]						      =    torch.tensor(
            [2.3_804e-03, 2.0_752e-03, 1.9_836e-03, 2.1_057e-03, 1.6_174e-03,
             3.0_518e-04, 9.1_553e-05, 3.3_569e-04, 9.7_656e-04, 1.8_311e-03,
             2.0_142e-03, 2.1_057e-03, 1.7_395e-03, 4.5_776e-04, -3.9_673e-04,
             4.5_776e-04, 1.0_071e-03, 9.1_553e-05, 4.8_828e-04, 1.1_597e-03,
             7.3_242e-04, 9.4_604e-04, 1.8_005e-03, 1.8_311e-03, 8.8_501e-04,
             4.2_725e-04, 4.8_828e-04, 7.3_242e-04, 1.0_986e-03, 2.1_057e-03]     )
        # fmt: on
        a	:     int						      =    self._load_datasamples(1     )
        a	:     List[str]						      =    [np.tile(input_audio[0][None]				,  reps=(2, 1)     )]
        input_audio[0][1] *= 0.5
        a	:     str						      =    EncodecFeatureExtractor(feature_size=2     )
        a	:     Any						      =    feature_extractor(_lowercase				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(input_values.shape				,  (1, 2, 9_36_80)     )
        self.assertTrue(torch.allclose(input_values[0, 0, :30]				,  _lowercase				,  atol=1e-6     )     )
        self.assertTrue(torch.allclose(input_values[0, 1, :30]				,  EXPECTED_INPUT_VALUES * 0.5				,  atol=1e-6     )     )


   def _lowercase(     self     )						->Union[str, Any]:
        a	:     Dict						      =    self._load_datasamples(2     )
        # would be easier if the stride was like
        a	:     Tuple						      =    EncodecFeatureExtractor(feature_size=1				,  chunk_length_s=1				,  overlap=0.01     )

        # pad and trunc raise an error ?
        with self.assertRaisesRegex(
            _lowercase				,  '''^Both padding and truncation were set. Make sure you only set one.$'''				,  ):
             a	:     str						      =    feature_extractor(
                 _lowercase				,  padding='''max_length'''				,  truncation=_lowercase				,  return_tensors='''pt'''     ).input_values

        # truncate to chunk
        a	:     Optional[int]						      =    feature_extractor(_lowercase				,  truncation=_lowercase				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(truncated_outputs.shape				,  (2, 1, 7_15_20)     )  # 2 chunks

        # force truncate to max_length
        a	:     Any						      =    feature_extractor(
            _lowercase				,  truncation=_lowercase				,  max_length=4_80_00				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(truncated_outputs.shape				,  (2, 1, 4_80_00)     )

        # pad to chunk
        a	:     Tuple						      =    feature_extractor(_lowercase				,  padding=_lowercase				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(padded_outputs.shape				,  (2, 1, 9_52_80)     )

        # pad to chunk
        a	:     Dict						      =    feature_extractor(_lowercase				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(truncated_outputs.shape				,  (2, 1, 9_52_80)     )

        # force pad to max length
        a	:     Optional[Any]						      =    feature_extractor(
            _lowercase				,  padding='''max_length'''				,  max_length=10_00_00				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(truncated_outputs.shape				,  (2, 1, 10_00_00)     )

        # force no pad
        with self.assertRaisesRegex(
            _lowercase				,  '''^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'''				,  ):
             a	:     Any						      =    feature_extractor(_lowercase				,  padding=_lowercase				,  return_tensors='''pt'''     ).input_values

        a	:     Tuple						      =    feature_extractor(input_audio[0]				,  padding=_lowercase				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(truncated_outputs.shape				,  (1, 1, 9_36_80)     )

        # no pad if no chunk_length_s
        a	:     Any						      =    None
        with self.assertRaisesRegex(
            _lowercase				,  '''^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'''				,  ):
             a	:     Any						      =    feature_extractor(_lowercase				,  padding=_lowercase				,  return_tensors='''pt'''     ).input_values

        a	:     List[str]						      =    feature_extractor(input_audio[0]				,  padding=_lowercase				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(truncated_outputs.shape				,  (1, 1, 9_36_80)     )

        # no pad if no overlap
        a	:     Optional[int]						      =    2
        a	:     str						      =    None
        with self.assertRaisesRegex(
            _lowercase				,  '''^Unable to create tensor, you should probably activate padding with \'padding=True\' to have batched tensors with the same length.$'''				,  ):
             a	:     List[str]						      =    feature_extractor(_lowercase				,  padding=_lowercase				,  return_tensors='''pt'''     ).input_values

        a	:     str						      =    feature_extractor(input_audio[0]				,  padding=_lowercase				,  return_tensors='''pt'''     ).input_values
        self.assertEquals(truncated_outputs.shape				,  (1, 1, 9_36_80)     )
"""simple docstring"""







import os
import warnings
from pathlib import Path
from typing import Optional, Union

from .dynamic_module_utils import custom_object_save
from .tokenization_utils_base import PreTrainedTokenizerBase
from .utils import PushToHubMixin, copy_func, direct_transformers_import, logging


UpperCamelCase_							:							Optional[int]				 =		logging.get_logger(__name__)

# Dynamically import the Transformers module to grab the attribute classes of the processor form their names.
UpperCamelCase_							:							Optional[Any]				 =		direct_transformers_import(Path(__file__).parent)


UpperCamelCase_							:							Optional[int]				 =		{
    """AutoTokenizer""": """PreTrainedTokenizerBase""",
    """AutoFeatureExtractor""": """FeatureExtractionMixin""",
    """AutoImageProcessor""": """ImageProcessingMixin""",
}


class       _lowercase		( __lowerCamelCase	):
				lowerCamelCase__							=							['feature_extractor', 'tokenizer']
				# Names need to be attr_class for attr in attributes
				lowerCamelCase__							=							None
				lowerCamelCase__							=							None
				lowerCamelCase__							=							None

				def __init__(  self							:     Any							,     *_lowercase							:     Optional[int]							,     **_lowercase							:     int  ):


										'''simple docstring'''
										# Sanitize args and kwargs
										for key in kwargs:
																if key not in self.attributes:
																						raise TypeError(f'Unexpected keyword argument {key}.'  )
										for arg, attribute_name in zip(_lowercase							,     self.attributes  ):
																if attribute_name in kwargs:
																						raise TypeError(f'Got multiple values for argument {attribute_name}.'  )
																else:
																						UpperCAmelCase__       						=		arg

										if len(_lowercase  ) != len(self.attributes  ):
																raise ValueError(
																    f'This processor requires {len(self.attributes  )} arguments: {", ".join(self.attributes  )}. Got '
																    f'{len(_lowercase  )} arguments instead.'  )

										# Check each arg is of the proper class (this will also catch a user initializing in the wrong order)
										for attribute_name, arg in kwargs.items():
																UpperCAmelCase__       						=		getattr(self							,     f'{attribute_name}_class'  )
																# Nothing is ever going to be an instance of "AutoXxx", in that case we check the base class.
																UpperCAmelCase__       						=		AUTO_TO_BASE_CLASS_MAPPING.get(_lowercase							,     _lowercase  )
																if isinstance(_lowercase							,     _lowercase  ):
																						UpperCAmelCase__       						=		tuple(getattr(_lowercase							,     _lowercase  ) for n in class_name if n is not None  )
																else:
																						UpperCAmelCase__       						=		getattr(_lowercase							,     _lowercase  )

																if not isinstance(_lowercase							,     _lowercase  ):
																						raise ValueError(
																						    f'Received a {type(_lowercase  ).__name__} for argument {attribute_name}, but a {class_name} was expected.'  )

																setattr(self							,     _lowercase							,     _lowercase  )

				def __repr__(  self							:     Tuple  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		[f'- {name}: {repr(getattr(self							,     _lowercase  )  )}' for name in self.attributes]
										UpperCAmelCase__       						=		"""\n""".join(_lowercase  )
										return f'{self.__class__.__name__}:\n{attributes_repr}'

				def 							A			(  self							:     Tuple							,     _lowercase							:     Dict							,     _lowercase							:     bool = False							,     **_lowercase							:     List[str]  ):


										'''simple docstring'''
										os.makedirs(_lowercase							,     exist_ok=_lowercase  )

										if push_to_hub:
																UpperCAmelCase__       						=		kwargs.pop("""commit_message"""							,     _lowercase  )
																UpperCAmelCase__       						=		kwargs.pop("""repo_id"""							,     save_directory.split(os.path.sep  )[-1]  )
																UpperCAmelCase__       						=		self._create_repo(_lowercase							,     **_lowercase  )
																UpperCAmelCase__       						=		self._get_files_timestamps(_lowercase  )
										# If we have a custom config, we copy the file defining it in the folder and set the attributes so it can be
										# loaded from the Hub.
										if self._auto_class is not None:
																UpperCAmelCase__       						=		[getattr(self							,     _lowercase  ) for attribute_name in self.attributes]
																UpperCAmelCase__       						=		[(a.init_kwargs if isinstance(_lowercase							,     _lowercase  ) else a) for a in attrs]
																custom_object_save(self							,     _lowercase							,     config=_lowercase  )

										for attribute_name in self.attributes:
																UpperCAmelCase__       						=		getattr(self							,     _lowercase  )
																# Include the processor class in the attribute config so this processor can then be reloaded with the
																# `AutoProcessor` API.
																if hasattr(_lowercase							,     """_set_processor_class"""  ):
																						attribute._set_processor_class(self.__class__.__name__  )
																attribute.save_pretrained(_lowercase  )

										if self._auto_class is not None:
																# We added an attribute to the init_kwargs of the tokenizers, which needs to be cleaned up.
																for attribute_name in self.attributes:
																						UpperCAmelCase__       						=		getattr(self							,     _lowercase  )
																						if isinstance(_lowercase							,     _lowercase  ):
																												del attribute.init_kwargs["auto_map"]

										if push_to_hub:
																self._upload_modified_files(
																    _lowercase							,     _lowercase							,     _lowercase							,     commit_message=_lowercase							,     token=kwargs.get("""use_auth_token"""  )							,     )

				@classmethod
				def 							A			(  cls							:     Dict							,     _lowercase							:     Union[str, os.PathLike]							,     _lowercase							:     Optional[Union[str, os.PathLike]] = None							,     _lowercase							:     bool = False							,     _lowercase							:     bool = False							,     _lowercase							:     Optional[Union[str, bool]] = None							,     _lowercase							:     str = "main"							,     **_lowercase							:     Dict							,     ):


										'''simple docstring'''
										UpperCAmelCase__       						=		cache_dir
										UpperCAmelCase__       						=		force_download
										UpperCAmelCase__       						=		local_files_only
										UpperCAmelCase__       						=		revision

										UpperCAmelCase__       						=		kwargs.pop("""use_auth_token"""							,     _lowercase  )
										if use_auth_token is not None:
																warnings.warn(
																    """The `use_auth_token` argument is deprecated and will be removed in v5 of Transformers."""							,     _lowercase  )
																if token is not None:
																						raise ValueError(
																						    """`token` and `use_auth_token` are both specified. Please set only the argument `token`."""  )
																UpperCAmelCase__       						=		use_auth_token

										if token is not None:
																# change to `token` in a follow-up PR
																UpperCAmelCase__       						=		token

										UpperCAmelCase__       						=		cls._get_arguments_from_pretrained(_lowercase							,     **_lowercase  )
										return cls(*_lowercase  )

				@classmethod
				def 							A			(  cls							:     List[Any]							,     _lowercase							:     Dict="AutoProcessor"  ):


										'''simple docstring'''
										if not isinstance(_lowercase							,     _lowercase  ):
																UpperCAmelCase__       						=		auto_class.__name__

										import transformers.models.auto as auto_module

										if not hasattr(_lowercase							,     _lowercase  ):
																raise ValueError(f'{auto_class} is not a valid auto class.'  )

										UpperCAmelCase__       						=		auto_class

				@classmethod
				def 							A			(  cls							:     Any							,     _lowercase							:     Tuple							,     **_lowercase							:     List[Any]  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		[]
										for attribute_name in cls.attributes:
																UpperCAmelCase__       						=		getattr(cls							,     f'{attribute_name}_class'  )
																if isinstance(_lowercase							,     _lowercase  ):
																						UpperCAmelCase__       						=		tuple(getattr(_lowercase							,     _lowercase  ) if n is not None else None for n in class_name  )
																						UpperCAmelCase__       						=		kwargs.get("""use_fast"""							,     _lowercase  )
																						if use_fast and classes[1] is not None:
																												UpperCAmelCase__       						=		classes[1]
																						else:
																												UpperCAmelCase__       						=		classes[0]
																else:
																						UpperCAmelCase__       						=		getattr(_lowercase							,     _lowercase  )

																args.append(attribute_class.from_pretrained(_lowercase							,     **_lowercase  )  )
										return args

				@property
				def 							A			(  self							:     int  ):


										'''simple docstring'''
										UpperCAmelCase__       						=		getattr(self							,     self.attributes[0]  )
										return getattr(_lowercase							,     """model_input_names"""							,     _lowercase  )
UpperCamelCase_							:							List[Any]				 =		copy_func(ProcessorMixin.push_to_hub)
if ProcessorMixin.push_to_hub.__doc__ is not None:
						UpperCamelCase_							:							Dict				 =		ProcessorMixin.push_to_hub.__doc__.format(
						    object="""processor""", object_class="""AutoProcessor""", object_files="""processor files"""
						)

