



"""simple docstring"""
import importlib.metadata
import warnings
from copy import deepcopy

from packaging import version

from ..utils import logging
from .import_utils import is_accelerate_available, is_bitsandbytes_available


if is_bitsandbytes_available():
					import bitsandbytes as bnb
					import torch
					import torch.nn as nn

					from ..pytorch_utils import ConvaD

if is_accelerate_available():
					from accelerate import init_empty_weights
					from accelerate.utils import find_tied_parameters

SCREAMING_SNAKE_CASE_				= logging.get_logger(__name__)



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None	)			->      List[Any]:
		# Recurse if needed
		if "." in tensor_name:
				a_     :      List[str] 						=     tensor_name.split("."	)
				for split in splits[:-1]:
						a_     :      Union[str, Any] 						=     getattr(SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)
						if new_module is None:
								raise ValueError(F"""{module} has no attribute {split}."""	)
						a_     :      int 						=     new_module
				a_     :      Any 						=     splits[-1]

		if tensor_name not in module._parameters and tensor_name not in module._buffers:
				raise ValueError(F"""{module} does not have a parameter or a buffer named {tensor_name}."""	)
		a_     :      Optional[int] 						=     tensor_name in module._buffers
		a_     :      Any 						=     getattr(SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)

		if old_value.device == torch.device("meta"	) and device not in ["meta", torch.device("meta"	)] and value is None:
				raise ValueError(F"""{tensor_name} is on the meta device, we need a `value` to put in on {device}."""	)

		a_     :      Optional[int] 						=     False
		a_     :      Tuple 						=     False
		if is_buffer or not is_bitsandbytes_available():
				a_     :      Union[str, Any] 						=     False
				a_     :      List[Any] 						=     False
		else:
				a_     :      Dict 						=     hasattr(bnb.nn, "Params4bit"	) and isinstance(module._parameters[tensor_name], bnb.nn.Paramsabit	)
				a_     :      Dict 						=     isinstance(module._parameters[tensor_name], bnb.nn.IntaParams	)

		if is_abit or is_abit:
				a_     :      Tuple 						=     module._parameters[tensor_name]
				if param.device.type != "cuda":
						if value is None:
								a_     :      int 						=     old_value.to(SCREAMING_SNAKE_CASE__	)
						elif isinstance(SCREAMING_SNAKE_CASE__, torch.Tensor	):
								a_     :      Any 						=     value.to("cpu"	)
								if value.dtype == torch.inta:
										a_     :      str 						=     version.parse(importlib.metadata.version("bitsandbytes"	)	) > version.parse(
										    "0.37.2"	)
										if not is_abit_serializable:
												raise ValueError(
												    "Detected int8 weights but the version of bitsandbytes is not compatible with int8 serialization. "
												    "Make sure to download the latest `bitsandbytes` version. `pip install --upgrade bitsandbytes`."	)
						else:
								a_     :      Dict 						=     torch.tensor(SCREAMING_SNAKE_CASE__, device="cpu"	)

						# Support models using `Conv1D` in place of `nn.Linear` (e.g. gpt2) by transposing the weight matrix prior to quantization.
						# Since weights are saved in the correct "orientation", we skip transposing when loading.
						if issubclass(module.source_cls, SCREAMING_SNAKE_CASE__	) and fpaa_statistics is None:
								a_     :      Optional[Any] 						=     new_value.T

						a_     :      Union[str, Any] 						=     old_value.__dict__
						if is_abit:
								a_     :      List[Any] 						=     bnb.nn.IntaParams(SCREAMING_SNAKE_CASE__, requires_grad=SCREAMING_SNAKE_CASE__, **SCREAMING_SNAKE_CASE__	).to(SCREAMING_SNAKE_CASE__	)
						elif is_abit:
								a_     :      Optional[Any] 						=     bnb.nn.Paramsabit(SCREAMING_SNAKE_CASE__, requires_grad=SCREAMING_SNAKE_CASE__, **SCREAMING_SNAKE_CASE__	).to(SCREAMING_SNAKE_CASE__	)

						a_     :      List[Any] 						=     new_value
						if fpaa_statistics is not None:
								setattr(module.weight, "SCB", fpaa_statistics.to(SCREAMING_SNAKE_CASE__	)	)

		else:
				if value is None:
						a_     :      str 						=     old_value.to(SCREAMING_SNAKE_CASE__	)
				elif isinstance(SCREAMING_SNAKE_CASE__, torch.Tensor	):
						a_     :      Optional[int] 						=     value.to(SCREAMING_SNAKE_CASE__	)
				else:
						a_     :      int 						=     torch.tensor(SCREAMING_SNAKE_CASE__, device=SCREAMING_SNAKE_CASE__	)

				if is_buffer:
						a_     :      int 						=     new_value
				else:
						a_     :      Optional[int] 						=     nn.Parameter(SCREAMING_SNAKE_CASE__, requires_grad=old_value.requires_grad	)
						a_     :      Any 						=     new_value



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=False	)			->      Optional[Any]:
		for name, module in model.named_children():
				if current_key_name is None:
						a_     :      Tuple 						=     []
				current_key_name.append(SCREAMING_SNAKE_CASE__	)

				if (isinstance(SCREAMING_SNAKE_CASE__, nn.Linear	) or isinstance(SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)) and name not in modules_to_not_convert:
						# Check if the current key is not in the `modules_to_not_convert`
						if not any(key in ".".join(SCREAMING_SNAKE_CASE__	) for key in modules_to_not_convert	):
								with init_empty_weights():
										if isinstance(SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	):
												a_						,				a_     :      Optional[Any] 						=     module.weight.shape
										else:
												a_     :      int 						=     module.in_features
												a_     :      Union[str, Any] 						=     module.out_features

										if quantization_config.quantization_method() == "llm_int8":
												a_     :      str 						=     bnb.nn.LinearabitLt(
												    SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, module.bias is not None, has_fpaa_weights=quantization_config.llm_inta_has_fpaa_weight, threshold=quantization_config.llm_inta_threshold, )
												a_     :      int 						=     True
										else:
												if (
												    quantization_config.llm_inta_skip_modules is not None
												    and name in quantization_config.llm_inta_skip_modules
												):
														pass
												else:
														a_     :      int 						=     bnb.nn.Linearabit(
														    SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, module.bias is not None, quantization_config.bnb_abit_compute_dtype, compress_statistics=quantization_config.bnb_abit_use_double_quant, quant_type=quantization_config.bnb_abit_quant_type, )
														a_     :      int 						=     True
                    # Store the module class in case we need to transpose the weight later
										a_     :      Union[str, Any] 						=     type(SCREAMING_SNAKE_CASE__	)
										# Force requires grad to False to avoid unexpected errors
										model._modules[name].requires_grad_(SCREAMING_SNAKE_CASE__	)
				if len(list(module.children()	)	) > 0:
						a_						,				a_     :      str 						=     _replace_with_bnb_linear(
						    SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, has_been_replaced=SCREAMING_SNAKE_CASE__, )
				# Remove the last key for recursion
				current_key_name.pop(-1	)
		return model, has_been_replaced



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None	)			->      Tuple:
		a_     :      List[str] 						=     ["lm_head"] if modules_to_not_convert is None else modules_to_not_convert
		a_						,				a_     :      Dict 						=     _replace_with_bnb_linear(
		    SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	)

		if not has_been_replaced:
				logger.warning(
				    "You are loading your model in 8bit or 4bit but no linear modules were found in your model."
				    " Please double check your model architecture, or submit an issue on github if you think this is"
				    " a bug."	)

		return model



def        lowerCAmelCase_						(							*SCREAMING_SNAKE_CASE__, **SCREAMING_SNAKE_CASE__	)			->      Tuple:
		warnings.warn(
		    "`replace_8bit_linear` will be deprecated in a future version, please use `replace_with_bnb_linear` instead", SCREAMING_SNAKE_CASE__, )
		return replace_with_bnb_linear(*SCREAMING_SNAKE_CASE__, **SCREAMING_SNAKE_CASE__	)



def        lowerCAmelCase_						(							*SCREAMING_SNAKE_CASE__, **SCREAMING_SNAKE_CASE__	)			->      Dict:
		warnings.warn(
		    "`set_module_8bit_tensor_to_device` will be deprecated in a future version, please use `set_module_quantized_tensor_to_device` instead", SCREAMING_SNAKE_CASE__, )
		return set_module_quantized_tensor_to_device(*SCREAMING_SNAKE_CASE__, **SCREAMING_SNAKE_CASE__	)



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__	)			->      Optional[int]:
		a_     :      Optional[Any] 						=     deepcopy(SCREAMING_SNAKE_CASE__	)  # this has 0 cost since it is done inside `init_empty_weights` context manager`
		tied_model.tie_weights()

		a_     :      int 						=     find_tied_parameters(SCREAMING_SNAKE_CASE__	)
		# For compatibility with Accelerate < 0.18
		if isinstance(SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__	):
				a_     :      List[Any] 						=     sum(list(tied_params.values()	), []	) + list(tied_params.keys()	)
		else:
				a_     :      int 						=     sum(SCREAMING_SNAKE_CASE__, []	)
		a_     :      Optional[Any] 						=     len(SCREAMING_SNAKE_CASE__	) > 0

		# Check if it is a base model
		a_     :      Optional[Any] 						=     not hasattr(SCREAMING_SNAKE_CASE__, model.base_model_prefix	)

		# Ignore this for base models (BertModel, GPT2Model, etc.)
		if (not has_tied_params) and is_base_model:
				return []

		# otherwise they have an attached head
		a_     :      Tuple 						=     list(model.named_children()	)
		a_     :      Dict 						=     [list_modules[-1][0]]

		# add last module together with tied weights
		a_     :      Tuple 						=     set(SCREAMING_SNAKE_CASE__	) - set(SCREAMING_SNAKE_CASE__	)
		a_     :      Any 						=     list(set(SCREAMING_SNAKE_CASE__	)	) + list(SCREAMING_SNAKE_CASE__	)

		# remove ".weight" from the keys
		a_     :      List[str] 						=     [".weight", ".bias"]
		a_     :      Optional[Any] 						=     []
		for name in list_untouched:
				for name_to_remove in names_to_remove:
						if name_to_remove in name:
								a_     :      int 						=     name.replace(SCREAMING_SNAKE_CASE__, ""	)
				filtered_module_names.append(SCREAMING_SNAKE_CASE__	)

		return filtered_module_names




"""simple docstring"""
import itertools
import os
import random
import tempfile
import unittest

import numpy as np

from transformers import TvltFeatureExtractor, is_datasets_available
from transformers.testing_utils import check_json_file_has_correct_format, require_torch, require_torchaudio
from transformers.utils.import_utils import is_torch_available

from ...test_sequence_feature_extraction_common import SequenceFeatureExtractionTestMixin


if is_torch_available():
					import torch

if is_datasets_available():
					from datasets import load_dataset

SCREAMING_SNAKE_CASE_				= random.Random()



def        lowerCAmelCase_						(							SCREAMING_SNAKE_CASE__, SCREAMING_SNAKE_CASE__=1.0, SCREAMING_SNAKE_CASE__=None, SCREAMING_SNAKE_CASE__=None	)			->      Tuple:
		if rng is None:
				a_     :      Any 						=     global_rng

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

		return values




class 			snake_case_						(						unittest.TestCase				):






							def __init__(   self			,     a_			,     a_=7			,     a_=4_0_0			,     a_=2_0_0_0			,     a_=2_0_4_8			,     a_=1_2_8			,     a_=1			,     a_=5_1_2			,     a_=3_0			,     a_=4_4_1_0_0			,     ):
									a_     :      Optional[Any] 						=     parent
									a_     :      Tuple 						=     batch_size
									a_     :      Union[str, Any] 						=     min_seq_length
									a_     :      Dict 						=     max_seq_length
									a_     :      Union[str, Any] 						=     (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
									a_     :      int 						=     spectrogram_length
									a_     :      Optional[Any] 						=     feature_size
									a_     :      Optional[int] 						=     num_audio_channels
									a_     :      List[str] 						=     hop_length
									a_     :      Optional[int] 						=     chunk_length
									a_     :      List[str] 						=     sampling_rate






							def snake_case_  (   self      ):
									return {
									    "spectrogram_length": self.spectrogram_length,
									    "feature_size": self.feature_size,
									    "num_audio_channels": self.num_audio_channels,
									    "hop_length": self.hop_length,
									    "chunk_length": self.chunk_length,
									    "sampling_rate": self.sampling_rate,
									}






							def snake_case_  (   self			,     a_=False			,     a_=False      ):
									def _flatten(a_      ):
											return list(itertools.chain(*a_      )      )

									if equal_length:
											a_     :      Tuple 						=     [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[Any] 						=     [
											    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(a_      ) for x in speech_inputs]
									return speech_inputs




@require_torch
@require_torchaudio
class 			snake_case_						(						a_     ,unittest.TestCase				):
							__lowerCAmelCase    					=					TvltFeatureExtractor






							def snake_case_  (   self      ):
									a_     :      Dict 						=     TvltFeatureExtractionTester(self      )






							def snake_case_  (   self      ):
									a_     :      Optional[Any] 						=     self.feature_extraction_class(**self.feat_extract_dict      )
									self.assertTrue(hasattr(a_			,     "spectrogram_length"      )      )
									self.assertTrue(hasattr(a_			,     "feature_size"      )      )
									self.assertTrue(hasattr(a_			,     "num_audio_channels"      )      )
									self.assertTrue(hasattr(a_			,     "hop_length"      )      )
									self.assertTrue(hasattr(a_			,     "chunk_length"      )      )
									self.assertTrue(hasattr(a_			,     "sampling_rate"      )      )






							def snake_case_  (   self      ):
									a_     :      List[Any] 						=     self.feature_extraction_class(**self.feat_extract_dict      )

									with tempfile.TemporaryDirectory() as tmpdirname:
											a_     :      List[Any] 						=     feat_extract_first.save_pretrained(a_      )[0]
											check_json_file_has_correct_format(a_      )
											a_     :      Tuple 						=     self.feature_extraction_class.from_pretrained(a_      )

									a_     :      Dict 						=     feat_extract_first.to_dict()
									a_     :      List[str] 						=     feat_extract_second.to_dict()
									a_     :      List[Any] 						=     dict_first.pop("mel_filters"      )
									a_     :      str 						=     dict_second.pop("mel_filters"      )
									self.assertTrue(np.allclose(a_			,     a_      )      )
									self.assertEqual(a_			,     a_      )






							def snake_case_  (   self      ):
									a_     :      List[str] 						=     self.feature_extraction_class(**self.feat_extract_dict      )

									with tempfile.TemporaryDirectory() as tmpdirname:
											a_     :      Dict 						=     os.path.join(a_			,     "feat_extract.json"      )
											feat_extract_first.to_json_file(a_      )
											a_     :      str 						=     self.feature_extraction_class.from_json_file(a_      )

									a_     :      Dict 						=     feat_extract_first.to_dict()
									a_     :      str 						=     feat_extract_second.to_dict()
									a_     :      int 						=     dict_first.pop("mel_filters"      )
									a_     :      List[Any] 						=     dict_second.pop("mel_filters"      )
									self.assertTrue(np.allclose(a_			,     a_      )      )
									self.assertEqual(a_			,     a_      )






							def snake_case_  (   self      ):
									# Initialize feature_extractor
									a_     :      str 						=     self.feature_extraction_class(**self.feat_extract_dict      )

									# create three inputs of length 800, 1000, and 1200
									a_     :      Optional[Any] 						=     [floats_list((1, x)      )[0] for x in range(8_0_0			,     1_4_0_0			,     2_0_0      )]
									a_     :      Union[str, Any] 						=     [np.asarray(a_      ) for speech_input in speech_inputs]

									# Test not batched input
									a_     :      Dict 						=     feature_extractor(np_speech_inputs[0]			,     return_tensors="np"			,     sampling_rate=4_4_1_0_0      ).audio_values

									self.assertTrue(encoded_audios.ndim == 4      )
									self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size      )
									self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length      )
									self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels      )

									# Test batched
									a_     :      int 						=     feature_extractor(a_			,     return_tensors="np"			,     sampling_rate=4_4_1_0_0      ).audio_values

									self.assertTrue(encoded_audios.ndim == 4      )
									self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size      )
									self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length      )
									self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels      )

									# Test audio masking
									a_     :      Optional[int] 						=     feature_extractor(
									    a_			,     return_tensors="np"			,     sampling_rate=4_4_1_0_0			,     mask_audio=a_      ).audio_values

									self.assertTrue(encoded_audios.ndim == 4      )
									self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size      )
									self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length      )
									self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels      )

									# Test 2-D numpy arrays are batched.
									a_     :      List[str] 						=     [floats_list((1, x)      )[0] for x in (8_0_0, 8_0_0, 8_0_0)]
									a_     :      Union[str, Any] 						=     np.asarray(a_      )
									a_     :      Dict 						=     feature_extractor(a_			,     return_tensors="np"			,     sampling_rate=4_4_1_0_0      ).audio_values
									self.assertTrue(encoded_audios.ndim == 4      )
									self.assertTrue(encoded_audios.shape[-1] == feature_extractor.feature_size      )
									self.assertTrue(encoded_audios.shape[-2] <= feature_extractor.spectrogram_length      )
									self.assertTrue(encoded_audios.shape[-3] == feature_extractor.num_channels      )






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

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






							def snake_case_  (   self      ):
									a_     :      List[str] 						=     self._load_datasamples(1      )
									a_     :      Any 						=     TvltFeatureExtractor()
									a_     :      Union[str, Any] 						=     feature_extractor(a_			,     return_tensors="pt"      ).audio_values

									self.assertEquals(audio_values.shape			,     (1, 1, 1_9_2, 1_2_8)      )

									a_     :      Dict 						=     torch.tensor([[-0.3_032, -0.2_708], [-0.4_434, -0.4_007]]      )
									self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2]			,     a_			,     atol=1e-4      )      )