
'''simple docstring'''
import datasets

from .evaluate import (
    apply_no_ans_threshold,
    find_all_best_thresh,
    get_raw_scores,
    make_eval_dict,
    make_qid_to_has_ans,
    merge_eval,
)


lowercase__		    =	"\\n@inproceedings{Rajpurkar2016SQuAD10,\n  title={SQuAD: 100, 000+ Questions for Machine Comprehension of Text},\n  author={Pranav Rajpurkar and Jian Zhang and Konstantin Lopyrev and Percy Liang},\n  booktitle={EMNLP},\n  year={2016}\n}\n"

lowercase__		    =	"\nThis metric wrap the official scoring script for version 2 of the Stanford Question\nAnswering Dataset (SQuAD).\n\nStanford Question Answering Dataset (SQuAD) is a reading comprehension dataset, consisting of questions posed by\ncrowdworkers on a set of Wikipedia articles, where the answer to every question is a segment of text, or span,\nfrom the corresponding reading passage, or the question might be unanswerable.\n\nSQuAD2.0 combines the 100,000 questions in SQuAD1.1 with over 50,000 unanswerable questions\nwritten adversarially by crowdworkers to look similar to answerable ones.\nTo do well on SQuAD2.0, systems must not only answer questions when possible, but also\ndetermine when no answer is supported by the paragraph and abstain from answering.\n"

lowercase__		    =	"\nComputes SQuAD v2 scores (F1 and EM).\nArgs:\n    predictions: List of triple for question-answers to score with the following elements:\n        - the question-answer 'id' field as given in the references (see below)\n        - the text of the answer\n        - the probability that the question has no answer\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 list of Dict {'text': text of the answer as a string}\n    no_answer_threshold: float\n        Probability threshold to decide that a question has no answer.\nReturns:\n    'exact': Exact match (the normalized answer exactly match the gold answer)\n    'f1': The F-score of predicted tokens versus the gold answer\n    'total': Number of score considered\n    'HasAns_exact': Exact match (the normalized answer exactly match the gold answer)\n    'HasAns_f1': The F-score of predicted tokens versus the gold answer\n    'HasAns_total': Number of score considered\n    'NoAns_exact': Exact match (the normalized answer exactly match the gold answer)\n    'NoAns_f1': The F-score of predicted tokens versus the gold answer\n    'NoAns_total': Number of score considered\n    'best_exact': Best exact match (with varying threshold)\n    'best_exact_thresh': No-answer probability threshold associated to the best exact match\n    'best_f1': Best F1 (with varying threshold)\n    'best_f1_thresh': No-answer probability threshold associated to the best F1\nExamples:\n\n    >>> predictions = [{'prediction_text': '1976', 'id': '56e10a3be3433e1400422b22', 'no_answer_probability': 0.}]\n    >>> references = [{'answers': {'answer_start': [97], 'text': ['1976']}, 'id': '56e10a3be3433e1400422b22'}]\n    >>> squad_v2_metric = datasets.load_metric(\"squad_v2\")\n    >>> results = squad_v2_metric.compute(predictions=predictions, references=references)\n    >>> print(results)\n    {'exact': 100.0, 'f1': 100.0, 'total': 1, 'HasAns_exact': 100.0, 'HasAns_f1': 100.0, 'HasAns_total': 1, 'best_exact': 100.0, 'best_exact_thresh': 0.0, 'best_f1': 100.0, 'best_f1_thresh': 0.0}\n"






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







		"""simple docstring"""

		def 	lowerCAmelCase							(	self							:       List[Any]	)		->       Tuple:






				"""simple docstring"""







				return datasets.MetricInfo(
				    description=_DESCRIPTION   ,     citation=_CITATION   ,     inputs_description=_KWARGS_DESCRIPTION   ,     features=datasets.Features(
				        {
				            '''predictions''': {
				                '''id''': datasets.Value('''string'''	),
				                '''prediction_text''': datasets.Value('''string'''	),
				                '''no_answer_probability''': datasets.Value('''float32'''	),
				            },
				            '''references''': {
				                '''id''': datasets.Value('''string'''	),
				                '''answers''': datasets.features.Sequence(
				                    {'''text''': datasets.Value('''string'''	), '''answer_start''': datasets.Value('''int32'''	)}	),
				            },
				        }	)   ,     codebase_urls=['''https://rajpurkar.github.io/SQuAD-explorer/''']   ,     reference_urls=['''https://rajpurkar.github.io/SQuAD-explorer/''']   ,     )



		def 	lowerCAmelCase							(	self							:       str   ,     UpperCamelCase__							:       Tuple   ,     UpperCamelCase__							:       Dict   ,     UpperCamelCase__							:       List[str]=1.0	)		->       str:






				"""simple docstring"""







				snake_case   :    str			       =		{p['''id''']: p['''no_answer_probability'''] for p in predictions}
				snake_case   :    Union[str, Any]			       =		[{'''paragraphs''': [{'''qas''': references}]}]
				snake_case   :    int			       =		{p['''id''']: p['''prediction_text'''] for p in predictions}

				snake_case   :    Optional[int]			       =		make_qid_to_has_ans(UpperCamelCase__	)  # maps qid to True/False
				snake_case   :    Optional[int]			       =		[k for k, v in qid_to_has_ans.items() if v]
				snake_case   :    Dict			       =		[k for k, v in qid_to_has_ans.items() if not v]

				snake_case       ,snake_case   :    List[Any]			       =		get_raw_scores(UpperCamelCase__   ,     UpperCamelCase__	)
				snake_case   :    List[Any]			       =		apply_no_ans_threshold(UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__	)
				snake_case   :    List[Any]			       =		apply_no_ans_threshold(UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__	)
				snake_case   :    int			       =		make_eval_dict(UpperCamelCase__   ,     UpperCamelCase__	)

				if has_ans_qids:
						snake_case   :    Any			       =		make_eval_dict(UpperCamelCase__   ,     UpperCamelCase__   ,     qid_list=UpperCamelCase__	)
						merge_eval(UpperCamelCase__   ,     UpperCamelCase__   ,     '''HasAns'''	)
				if no_ans_qids:
						snake_case   :    Any			       =		make_eval_dict(UpperCamelCase__   ,     UpperCamelCase__   ,     qid_list=UpperCamelCase__	)
						merge_eval(UpperCamelCase__   ,     UpperCamelCase__   ,     '''NoAns'''	)
				find_all_best_thresh(UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__	)
				return dict(UpperCamelCase__	)




'''simple docstring'''
import argparse
import os
from dataclasses import dataclass

import torch
import torchaudio
from tqdm import tqdm

from transformers import AutoConfig, AutoModelForCTC, AutoProcessor






class      snake_case__    :







		"""simple docstring"""

		def __init__(	self							:       str   ,     UpperCamelCase__							:       List[str]   ,     UpperCamelCase__							:       Any   ,     UpperCamelCase__							:       Dict	)		->       List[Any]:






				"""simple docstring"""







				snake_case   :    Union[str, Any]			       =		cuda
				snake_case   :    int			       =		AutoConfig.from_pretrained(UpperCamelCase__	)
				snake_case   :    int			       =		AutoModelForCTC.from_pretrained(UpperCamelCase__	)
				self.model.eval()
				if self.cuda:
						self.model.to(device='''cuda'''	)
				snake_case   :    Tuple			       =		AutoProcessor.from_pretrained(UpperCamelCase__	)
				snake_case   :    str			       =		torchaudio.transforms.Resample(UpperCamelCase__   ,     1_6000	)
				snake_case   :    Any			       =		0
				snake_case   :    int			       =		list(self.processor.tokenizer.get_vocab().keys()	)
				for i in range(len(UpperCamelCase__	)	):
						if vocab[i] == "[PAD]" or vocab[i] == "<pad>":
								snake_case   :    Union[str, Any]			       =		i
				print('''Blank Token id [PAD]/<pad>'''   ,     UpperCamelCase__	)
				snake_case   :    List[Any]			       =		blank_id



		def 	lowerCAmelCase							(	self							:       Union[str, Any]   ,     UpperCamelCase__							:       str	)		->       List[str]:






				"""simple docstring"""







				snake_case       ,snake_case   :    Tuple			       =		torchaudio.load(UpperCamelCase__	)
				snake_case   :    Optional[Any]			       =		self.resampler(UpperCamelCase__	).squeeze().numpy()
				return speech



		def 	lowerCAmelCase							(	self							:       Optional[Any]   ,     UpperCamelCase__							:       int	)		->       Dict:






				"""simple docstring"""







				snake_case   :    List[str]			       =		self.blank_id
				snake_case   :    Union[str, Any]			       =		'''|'''.join(item['''sent'''].split(''' '''	)	)
				if not os.path.isfile(item['''wav_path''']	):
						print(item['''wav_path''']   ,     '''not found in wavs directory'''	)

				snake_case   :    List[Any]			       =		self.speech_file_to_array_fn(item['''wav_path''']	)
				snake_case   :    List[str]			       =		self.processor(UpperCamelCase__   ,     sampling_rate=1_6000   ,     return_tensors='''pt'''   ,     padding=UpperCamelCase__	)
				if self.cuda:
						snake_case   :    str			       =		inputs.to(device='''cuda'''	)

				with torch.no_grad():
						snake_case   :    List[Any]			       =		self.model(inputs.input_values	).logits

				# get the emission probability at frame level
				snake_case   :    str			       =		torch.log_softmax(UpperCamelCase__   ,     dim=-1	)
				snake_case   :    Optional[Any]			       =		emissions[0].cpu().detach()

				# get labels from vocab
				snake_case   :    Dict			       =		([''''''] + list(self.processor.tokenizer.get_vocab().keys()	))[
				    :-1
				]  # logits don't align with the tokenizer's vocab

				snake_case   :    List[str]			       =		{c: i for i, c in enumerate(UpperCamelCase__	)}
				snake_case   :    Any			       =		[]
				for c in transcript:
						if c in dictionary:
								tokens.append(dictionary[c]	)

				def get_trellis(UpperCamelCase__							:       str   ,     UpperCamelCase__							:       Dict   ,     UpperCamelCase__							:       str=0	):
						snake_case   :    List[str]			       =		emission.size(0	)
						snake_case   :    List[str]			       =		len(UpperCamelCase__	)

						# Trellis has extra diemsions for both time axis and tokens.
						# The extra dim for tokens represents <SoS> (start-of-sentence)
						# The extra dim for time axis is for simplification of the code.
						snake_case   :    List[Any]			       =		torch.full((num_frames + 1, num_tokens + 1)   ,     -float('''inf'''	)	)
						snake_case   :    Optional[int]			       =		0
						for t in range(UpperCamelCase__	):
								snake_case   :    Any			       =		torch.maximum(
								    # Score for staying at the same token
								    trellis[t, 1:] + emission[t, blank_id]   ,     trellis[t, :-1] + emission[t, tokens]   ,     )
						return trellis

				snake_case   :    Optional[Any]			       =		get_trellis(UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__	)






				@dataclass
				class      snake_case__    :







						"""simple docstring"""

						lowerCamelCase							=		42
						lowerCamelCase							=		42
						lowerCamelCase							=		42

				def backtrack(UpperCamelCase__							:       int   ,     UpperCamelCase__							:       Optional[Any]   ,     UpperCamelCase__							:       Optional[Any]   ,     UpperCamelCase__							:       Union[str, Any]=0	):
						snake_case   :    List[str]			       =		trellis.size(1	) - 1
						snake_case   :    Any			       =		torch.argmax(trellis[:, j]	).item()

						snake_case   :    Dict			       =		[]
						for t in range(UpperCamelCase__   ,     0   ,     -1	):
								# 1. Figure out if the current position was stay or change
								# Note (again):
								# `emission[J-1]` is the emission at time frame `J` of trellis dimension.
								# Score for token staying the same from time frame J-1 to T.
								snake_case   :    Union[str, Any]			       =		trellis[t - 1, j] + emission[t - 1, blank_id]
								# Score for token changing from C-1 at T-1 to J at T.
								snake_case   :    Optional[int]			       =		trellis[t - 1, j - 1] + emission[t - 1, tokens[j - 1]]

								# 2. Store the path with frame-wise probability.
								snake_case   :    Optional[Any]			       =		emission[t - 1, tokens[j - 1] if changed > stayed else 0].exp().item()
								# Return token index and time index in non-trellis coordinate.
								path.append(Point(j - 1   ,     t - 1   ,     UpperCamelCase__	)	)

								# 3. Update the token
								if changed > stayed:
										j -= 1
										if j == 0:
												break
						else:
								raise ValueError('''Failed to align'''	)
						return path[::-1]

				snake_case   :    Any			       =		backtrack(UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__   ,     UpperCamelCase__	)






				@dataclass
				class      snake_case__    :







						"""simple docstring"""

						lowerCamelCase							=		42
						lowerCamelCase							=		42
						lowerCamelCase							=		42
						lowerCamelCase							=		42



						def __repr__(	self							:       Dict	)		->       int:






								"""simple docstring"""







								return f'{self.label}\t{self.score:4.2f}\t{self.start*20:5d}\t{self.end*20:5d}'



						@property
						def 	lowerCAmelCase							(	self							:       List[Any]	)		->       Any:






								"""simple docstring"""







								return self.end - self.start

				def merge_repeats(UpperCamelCase__							:       str	):
						snake_case       ,snake_case   :    Union[str, Any]			       =		0, 0
						snake_case   :    List[Any]			       =		[]
						while ia < len(UpperCamelCase__	):
								while ia < len(UpperCamelCase__	) and path[ia].token_index == path[ia].token_index:
										ia += 1
								snake_case   :    Dict			       =		sum(path[k].score for k in range(UpperCamelCase__   ,     UpperCamelCase__	)	) / (ia - ia)
								segments.append(
								    Segment(
								        transcript[path[ia].token_index]   ,     path[ia].time_index   ,     path[ia - 1].time_index + 1   ,     UpperCamelCase__   ,     )	)
								snake_case   :    int			       =		ia
						return segments

				snake_case   :    Any			       =		merge_repeats(UpperCamelCase__	)
				with open(item['''out_path''']   ,     '''w'''	) as out_align:
						for seg in segments:
								out_align.write(str(UpperCamelCase__	) + '''\n'''	)



		def 	lowerCAmelCase							(	self							:       Dict   ,     UpperCamelCase__							:       Dict   ,     UpperCamelCase__							:       Union[str, Any]   ,     UpperCamelCase__							:       List[str]	)		->       List[str]:






				"""simple docstring"""







				if not os.path.exists(UpperCamelCase__	):
						os.makedirs(UpperCamelCase__	)

				# load text file
				snake_case   :    List[Any]			       =		open(UpperCamelCase__   ,     encoding='''utf8'''	).readlines()

				snake_case   :    Optional[int]			       =		[]
				for line in lines:
						if len(line.strip().split('''\t'''	)	) != 2:
								print('''Script must be in format: 00001  this is my sentence'''	)
								exit()

						snake_case       ,snake_case   :    Dict			       =		line.strip().split('''\t'''	)
						snake_case   :    Tuple			       =		os.path.join(UpperCamelCase__   ,     wav_name + '''.wav'''	)
						snake_case   :    str			       =		os.path.join(UpperCamelCase__   ,     wav_name + '''.txt'''	)

						items.append({'''sent''': sentence, '''wav_path''': wav_path, '''out_path''': out_path}	)
				print('''Number of samples found in script file'''   ,     len(UpperCamelCase__	)	)

				for item in tqdm(UpperCamelCase__	):
						self.align_single_sample(UpperCamelCase__	)







def 					_UpperCamelCase		(    )   ->				List[Any]:


		'''simple docstring'''


		snake_case   :    Optional[int]			       =		argparse.ArgumentParser()

		parser.add_argument(
		    '''--model_name'''      ,    type=SCREAMING_SNAKE_CASE__      ,    default='''arijitx/wav2vec2-xls-r-300m-bengali'''      ,    help='''wav2vec model name'''       )
		parser.add_argument('''--wav_dir'''      ,    type=SCREAMING_SNAKE_CASE__      ,    default='''./wavs'''      ,    help='''directory containing wavs'''       )
		parser.add_argument('''--text_file'''      ,    type=SCREAMING_SNAKE_CASE__      ,    default='''script.txt'''      ,    help='''file containing text'''       )
		parser.add_argument('''--input_wavs_sr'''      ,    type=SCREAMING_SNAKE_CASE__      ,    default=1_6000      ,    help='''sampling rate of input audios'''       )
		parser.add_argument(
		    '''--output_dir'''      ,    type=SCREAMING_SNAKE_CASE__      ,    default='''./out_alignment'''      ,    help='''output directory containing the alignment files'''       )
		parser.add_argument('''--cuda'''      ,    action='''store_true'''       )

		snake_case   :    Tuple			       =		parser.parse_args()

		snake_case   :    Tuple			       =		WavaVecaAligner(args.model_name      ,    args.input_wavs_sr      ,    args.cuda       )
		aligner.align_data(args.wav_dir      ,    args.text_file      ,    args.output_dir       )


if __name__ == "__main__":
						main()


