





from math import ceil
from typing import List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, spectrogram, window_function
from ...feature_extraction_sequence_utils import BatchFeature, SequenceFeatureExtractor
from ...utils import TensorType, logging


__lowerCAmelCase :List[Any]     =    logging.get_logger(__name__)






class        _a(    __A    ):
		lowerCamelCase__     :Tuple   				=		['audio_values', 'audio_mask']







		def __init__(    self ,						__snake_case=2_0_4_8 ,						__snake_case=1 ,						__snake_case=[1_6, 1_6] ,						__snake_case=1_2_8 ,						__snake_case=4_4_1_0_0 ,						__snake_case=8_6 ,						__snake_case=2_0_4_8 ,						__snake_case=0.0 ,						**__snake_case ,						)      ->      List[Any]:


							'''simple docstring'''

							super().__init__(
							    feature_size=__snake_case ,						sampling_rate=__snake_case ,						padding_value=__snake_case ,						**__snake_case ,						)

							_snake_case       :					Union[str, Any]				       =   spectrogram_length
							_snake_case       :					Optional[Any]				       =   num_channels
							_snake_case       :					str				       =   patch_size
							_snake_case       :					Dict				       =   feature_size // self.patch_size[1]
							_snake_case       :					Dict				       =   n_fft
							_snake_case       :					List[str]				       =   sampling_rate // hop_length_to_sampling_rate
							_snake_case       :					List[Any]				       =   sampling_rate
							_snake_case       :					List[str]				       =   padding_value
							_snake_case       :					Union[str, Any]				       =   mel_filter_bank(
							    num_frequency_bins=1 + n_fft // 2 ,						num_mel_filters=__snake_case ,						min_frequency=0.0 ,						max_frequency=2_20_50.0 ,						sampling_rate=__snake_case ,						norm="slaney" ,						mel_scale="slaney" ,						).T







		def        lowercase       (    self ,						__snake_case				)      ->      np.ndarray:


							'''simple docstring'''

							_snake_case       :					Optional[Any]				       =   spectrogram(
							    __snake_case ,						window_function(self.n_fft ,						"hann"				) ,						frame_length=self.n_fft ,						hop_length=self.hop_length ,						power=2.0 ,						mel_filters=self.mel_filters.T ,						log_mel="dB" ,						db_range=80.0 ,						)
							_snake_case       :					Optional[int]				       =   log_spec[:, :-1]
							_snake_case       :					int				       =   log_spec - 20.0
							_snake_case       :					str				       =   np.clip(log_spec / 40.0 ,						-2.0 ,						0.0				) + 1.0
							return log_spec







		def __call__(    self ,						__snake_case ,						__snake_case = None ,						__snake_case = True ,						__snake_case = None ,						__snake_case = False ,						__snake_case = False ,						**__snake_case ,						)      ->      BatchFeature:


							'''simple docstring'''

							if sampling_rate is not None:
												if sampling_rate != self.sampling_rate:
																	raise ValueError(
																	    "This feature extractor is set to support sampling rate"
																	    f""" of {self.sampling_rate}. Please make sure that the provided `raw_speech` input was sampled"""
																	    f""" with {self.sampling_rate} and not {sampling_rate}."""				)
							else:
												logger.warning(
												    "It is strongly recommended to pass the `sampling_rate` argument to this function. "
												    "Failing to do so can result in silent errors that might be hard to debug."				)

							_snake_case       :					Optional[Any]				       =   isinstance(__snake_case ,						np.ndarray				) and len(raw_speech.shape				) > 1
							if is_batched_numpy and len(raw_speech.shape				) > 2:
												raise ValueError(f"""Only mono-channel audio is supported for input to {self}"""				)
							_snake_case       :					str				       =   is_batched_numpy or (
							    isinstance(__snake_case ,						(list, tuple)				) and (isinstance(raw_speech[0] ,						(np.ndarray, tuple, list)				))
							)
							if is_batched:
												_snake_case       :					List[Any]				       =   [np.asarray([speech] ,						dtype=np.floataa				).T for speech in raw_speech]
							elif not is_batched and not isinstance(__snake_case ,						np.ndarray				):
												_snake_case       :					Union[str, Any]				       =   np.asarray(__snake_case ,						dtype=np.floataa				)
							elif isinstance(__snake_case ,						np.ndarray				) and raw_speech.dtype is np.dtype(np.floataa				):
												_snake_case       :					List[str]				       =   raw_speech.astype(np.floataa				)
							# always return batch
							if not is_batched:
												_snake_case       :					str				       =   [np.asarray([raw_speech]				).T]

							# Convert audio signals to log mel spectrograms, truncate by time axis
							_snake_case       :					List[Any]				       =   [
							    self._np_extract_fbank_features(waveform.squeeze()				).T[: self.spectrogram_length] for waveform in raw_speech
							]
							if isinstance(audio_features[0] ,						__snake_case				):
												_snake_case       :					Optional[Any]				       =   [np.asarray(__snake_case ,						dtype=np.floataa				) for feature in audio_features]

							# Create audio attention mask
							_snake_case       :					Tuple				       =   max(
							    [ceil(feature.shape[0] / self.patch_size[0]				) * self.freq_len for feature in audio_features]				)  # The maximum number of audio patches in a batch
							if return_attention_mask:
												_snake_case       :					Tuple				       =   [
												    (ceil(feature.shape[0] / self.patch_size[0]				) * self.freq_len) * [1]
												    + (max_patch_len - ceil(feature.shape[0] / self.patch_size[0]				) * self.freq_len) * [0]
												    for feature in audio_features
												]
												_snake_case       :					int				       =   np.array(__snake_case				).astype(np.floataa				)

							# convert into correct format for padding
							_snake_case       :					Optional[Any]				       =   max_patch_len // self.freq_len * self.patch_size[0]  # The maximum audio size in a batch
							_snake_case       :					int				       =   np.ones([len(__snake_case				), 1, max_time_len, self.feature_size]				).astype(np.floataa				)
							_snake_case       :					Union[str, Any]				       =   padded_audio_features * self.padding_value
							for i in range(len(__snake_case				)				):
												_snake_case       :					int				       =   audio_features[i]
												_snake_case       :					List[str]				       =   feature

							# return as BatchFeature
							if return_attention_mask:
												_snake_case       :					Union[str, Any]				       =   {"audio_values": padded_audio_features, "audio_mask": audio_mask}
							else:
												_snake_case       :					Dict				       =   {"audio_values": padded_audio_features}

							_snake_case       :					str				       =   BatchFeature(data=__snake_case ,						tensor_type=__snake_case				)
							return encoded_inputs






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