





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

__lowerCAmelCase :Tuple     =    random.Random()







def    A				(							UpperCAmelCase  ,  UpperCAmelCase=1.0  ,  UpperCAmelCase=None  ,  UpperCAmelCase=None				):
					if rng is None:
										_snake_case       :					Optional[int]				       =   global_rng

					_snake_case       :					Optional[Any]				       =   []
					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    ):







		def __init__(    self ,						__snake_case ,						__snake_case=7 ,						__snake_case=4_0_0 ,						__snake_case=2_0_0_0 ,						__snake_case=2_0_4_8 ,						__snake_case=1_2_8 ,						__snake_case=1 ,						__snake_case=5_1_2 ,						__snake_case=3_0 ,						__snake_case=4_4_1_0_0 ,						)      ->      str:


							'''simple docstring'''

							_snake_case       :					Optional[int]				       =   parent
							_snake_case       :					Tuple				       =   batch_size
							_snake_case       :					Tuple				       =   min_seq_length
							_snake_case       :					List[Any]				       =   max_seq_length
							_snake_case       :					Tuple				       =   (self.max_seq_length - self.min_seq_length) // (self.batch_size - 1)
							_snake_case       :					Any				       =   spectrogram_length
							_snake_case       :					Optional[Any]				       =   feature_size
							_snake_case       :					List[str]				       =   num_audio_channels
							_snake_case       :					Dict				       =   hop_length
							_snake_case       :					Union[str, Any]				       =   chunk_length
							_snake_case       :					Union[str, Any]				       =   sampling_rate







		def        lowercase       (    self				)      ->      int:


							'''simple docstring'''

							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        lowercase       (    self ,						__snake_case=False ,						__snake_case=False				)      ->      Union[str, Any]:


							'''simple docstring'''

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

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






@require_torch
@require_torchaudio
class        _a(    __A	,   unittest.TestCase    ):
		lowerCamelCase__     :str   				=		TvltFeatureExtractor







		def        lowercase       (    self				)      ->      Tuple:


							'''simple docstring'''

							_snake_case       :					int				       =   TvltFeatureExtractionTester(self				)







		def        lowercase       (    self				)      ->      Dict:


							'''simple docstring'''

							_snake_case       :					List[Any]				       =   self.feature_extraction_class(**self.feat_extract_dict				)
							self.assertTrue(hasattr(__snake_case ,						"spectrogram_length"				)				)
							self.assertTrue(hasattr(__snake_case ,						"feature_size"				)				)
							self.assertTrue(hasattr(__snake_case ,						"num_audio_channels"				)				)
							self.assertTrue(hasattr(__snake_case ,						"hop_length"				)				)
							self.assertTrue(hasattr(__snake_case ,						"chunk_length"				)				)
							self.assertTrue(hasattr(__snake_case ,						"sampling_rate"				)				)







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


							'''simple docstring'''

							_snake_case       :					List[str]				       =   self.feature_extraction_class(**self.feat_extract_dict				)

							with tempfile.TemporaryDirectory() as tmpdirname:
												_snake_case       :					Optional[int]				       =   feat_extract_first.save_pretrained(__snake_case				)[0]
												check_json_file_has_correct_format(__snake_case				)
												_snake_case       :					List[str]				       =   self.feature_extraction_class.from_pretrained(__snake_case				)

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







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


							'''simple docstring'''

							_snake_case       :					List[str]				       =   self.feature_extraction_class(**self.feat_extract_dict				)

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

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







		def        lowercase       (    self				)      ->      int:


							'''simple docstring'''

							_snake_case       :					Optional[int]				       =   self.feature_extraction_class(**self.feat_extract_dict				)

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

							# Test not batched input
							_snake_case       :					List[str]				       =   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
							_snake_case       :					Tuple				       =   feature_extractor(__snake_case ,						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
							_snake_case       :					int				       =   feature_extractor(
							    __snake_case ,						return_tensors="np" ,						sampling_rate=4_4_1_0_0 ,						mask_audio=__snake_case				).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.
							_snake_case       :					str				       =   [floats_list((1, x)				)[0] for x in (8_0_0, 8_0_0, 8_0_0)]
							_snake_case       :					Optional[Any]				       =   np.asarray(__snake_case				)
							_snake_case       :					Tuple				       =   feature_extractor(__snake_case ,						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        lowercase       (    self ,						__snake_case				)      ->      Optional[int]:


							'''simple docstring'''

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

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







		def        lowercase       (    self				)      ->      Dict:


							'''simple docstring'''

							_snake_case       :					Union[str, Any]				       =   self._load_datasamples(1				)
							_snake_case       :					List[Any]				       =   TvltFeatureExtractor()
							_snake_case       :					int				       =   feature_extractor(__snake_case ,						return_tensors="pt"				).audio_values

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

							_snake_case       :					Optional[int]				       =   torch.tensor([[-0.30_32, -0.27_08], [-0.44_34, -0.40_07]]				)
							self.assertTrue(torch.allclose(audio_values[0, 0, :2, :2] ,						__snake_case ,						atol=1E-4				)				)






import datasets

from .evaluate import evaluate


__lowerCAmelCase :Any     =    '\\n@article{hendrycks2021cuad,\n      title={CUAD: An Expert-Annotated NLP Dataset for Legal Contract Review},\n      author={Dan Hendrycks and Collin Burns and Anya Chen and Spencer Ball},\n      journal={arXiv preprint arXiv:2103.06268},\n      year={2021}\n}\n'

__lowerCAmelCase :Union[str, Any]     =    '\nThis metric wrap the official scoring script for version 1 of the Contract\nUnderstanding Atticus Dataset (CUAD).\nContract Understanding Atticus Dataset (CUAD) v1 is a corpus of more than 13,000 labels in 510\ncommercial legal contracts that have been manually labeled to identify 41 categories of important\nclauses that lawyers look for when reviewing contracts in connection with corporate transactions.\n'

__lowerCAmelCase :Optional[Any]     =    '\nComputes CUAD scores (EM, F1, AUPR, Precision@80%Recall, and Precision@90%Recall).\nArgs:\n    predictions: List of question-answers dictionaries with the following key-values:\n        - \'id\': id of the question-answer pair as given in the references (see below)\n        - \'prediction_text\': list of possible texts for the answer, as a list of strings\n        depending on a threshold on the confidence probability of each prediction.\n    references: List of question-answers dictionaries with the following key-values:\n        - \'id\': id of the question-answer pair (see above),\n        - \'answers\': a Dict in the CUAD dataset format\n            {\n                \'text\': list of possible texts for the answer, as a list of strings\n                \'answer_start\': list of start positions for the answer, as a list of ints\n            }\n            Note that answer_start values are not taken into account to compute the metric.\nReturns:\n    \'exact_match\': Exact match (the normalized answer exactly match the gold answer)\n    \'f1\': The F-score of predicted tokens versus the gold answer\n    \'aupr\': Area Under the Precision-Recall curve\n    \'prec_at_80_recall\': Precision at 80% recall\n    \'prec_at_90_recall\': Precision at 90% recall\nExamples:\n    >>> predictions = [{\'prediction_text\': [\'The seller:\', \'The buyer/End-User: Shenzhen LOHAS Supply Chain Management Co., Ltd.\'], \'id\': \'LohaCompanyltd_20191209_F-1_EX-10.16_11917878_EX-10.16_Supply Agreement__Parties\'}]\n    >>> references = [{\'answers\': {\'answer_start\': [143, 49], \'text\': [\'The seller:\', \'The buyer/End-User: Shenzhen LOHAS Supply Chain Management Co., Ltd.\']}, \'id\': \'LohaCompanyltd_20191209_F-1_EX-10.16_11917878_EX-10.16_Supply Agreement__Parties\'}]\n    >>> cuad_metric = datasets.load_metric("cuad")\n    >>> results = cuad_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {\'exact_match\': 100.0, \'f1\': 100.0, \'aupr\': 0.0, \'prec_at_80_recall\': 1.0, \'prec_at_90_recall\': 1.0}\n'






@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION	,   _KWARGS_DESCRIPTION    )
class        _a(    datasets.Metric    ):







		def        lowercase       (    self				)      ->      Optional[int]:


							'''simple docstring'''

							return datasets.MetricInfo(
							    description=_DESCRIPTION ,						citation=_CITATION ,						inputs_description=_KWARGS_DESCRIPTION ,						features=datasets.Features(
							        {
							            "predictions": {
							                "id": datasets.Value("string"				),
							                "prediction_text": datasets.features.Sequence(datasets.Value("string"				)				),
							            },
							            "references": {
							                "id": datasets.Value("string"				),
							                "answers": datasets.features.Sequence(
							                    {
							                        "text": datasets.Value("string"				),
							                        "answer_start": datasets.Value("int32"				),
							                    }				),
							            },
							        }				) ,						codebase_urls=["https://www.atticusprojectai.org/cuad"] ,						reference_urls=["https://www.atticusprojectai.org/cuad"] ,						)







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


							'''simple docstring'''

							_snake_case       :					str				       =   {prediction["id"]: prediction["prediction_text"] for prediction in predictions}
							_snake_case       :					Union[str, Any]				       =   [
							    {
							        "paragraphs": [
							            {
							                "qas": [
							                    {
							                        "answers": [{"text": answer_text} for answer_text in ref["answers"]["text"]],
							                        "id": ref["id"],
							                    }
							                    for ref in references
							                ]
							            }
							        ]
							    }
							]
							_snake_case       :					Union[str, Any]				       =   evaluate(dataset=__snake_case ,						predictions=__snake_case				)
							return score