


'''simple docstring'''







from __future__ import annotations

import unittest

from transformers import FunnelConfig, is_tf_available
from transformers.testing_utils import require_tf

from ...test_configuration_common import ConfigTester
from ...test_modeling_tf_common import TFModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_tf_available():
					import tensorflow as tf

					from transformers import (
					    TFFunnelBaseModel,
					    TFFunnelForMaskedLM,
					    TFFunnelForMultipleChoice,
					    TFFunnelForPreTraining,
					    TFFunnelForQuestionAnswering,
					    TFFunnelForSequenceClassification,
					    TFFunnelForTokenClassification,
					    TFFunnelModel,
					)


class 		__UpperCAmelCase   :


							def __init__( self					,     lowerCAmelCase_					,     lowerCAmelCase_=13					,     lowerCAmelCase_=7					,     lowerCAmelCase_=True					,     lowerCAmelCase_=True					,     lowerCAmelCase_=True					,     lowerCAmelCase_=True					,     lowerCAmelCase_=99					,     lowerCAmelCase_=[1, 1, 2]					,     lowerCAmelCase_=1					,     lowerCAmelCase_=32					,     lowerCAmelCase_=4					,     lowerCAmelCase_=8					,     lowerCAmelCase_=37					,     lowerCAmelCase_="gelu_new"					,     lowerCAmelCase_=0.1					,     lowerCAmelCase_=0.1					,     lowerCAmelCase_=0.0					,     lowerCAmelCase_=5_12					,     lowerCAmelCase_=3					,     lowerCAmelCase_=0.02					,     lowerCAmelCase_=3					,     lowerCAmelCase_=4					,     lowerCAmelCase_=None					,     lowerCAmelCase_=False					,     ):





								"""simple docstring"""






								_snake_case     		=	parent
								_snake_case     		=	batch_size
								_snake_case     		=	seq_length
								_snake_case     		=	is_training
								_snake_case     		=	use_input_mask
								_snake_case     		=	use_token_type_ids
								_snake_case     		=	use_labels
								_snake_case     		=	vocab_size
								_snake_case     		=	block_sizes
								_snake_case     		=	num_decoder_layers
								_snake_case     		=	d_model
								_snake_case     		=	n_head
								_snake_case     		=	d_head
								_snake_case     		=	d_inner
								_snake_case     		=	hidden_act
								_snake_case     		=	hidden_dropout
								_snake_case     		=	attention_dropout
								_snake_case     		=	activation_dropout
								_snake_case     		=	max_position_embeddings
								_snake_case     		=	type_vocab_size
								_snake_case     		=	2
								_snake_case     		=	num_labels
								_snake_case     		=	num_choices
								_snake_case     		=	scope
								_snake_case     		=	initializer_std

								# Used in the tests to check the size of the first attention layer
								_snake_case     		=	n_head
								# Used in the tests to check the size of the first hidden state
								_snake_case     		=	self.d_model
								# Used in the tests to check the number of output hidden states/attentions
								_snake_case     		=	sum(self.block_sizes						) + (0 if base else self.num_decoder_layers)
								# FunnelModel adds two hidden layers: input embeddings and the sum of the upsampled encoder hidden state with
								# the last hidden state of the first block (which is the first hidden state of the decoder).
								if not base:
									_snake_case     		=	self.num_hidden_layers + 2


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	ids_tensor([self.batch_size, self.seq_length]					,     self.vocab_size						)

								_snake_case     		=	None
								if self.use_input_mask:
									_snake_case     		=	random_attention_mask([self.batch_size, self.seq_length]						)

								_snake_case     		=	None
								if self.use_token_type_ids:
									_snake_case     		=	ids_tensor([self.batch_size, self.seq_length]					,     self.type_vocab_size						)

								_snake_case     		=	None
								_snake_case     		=	None
								_snake_case     		=	None
								if self.use_labels:
									_snake_case     		=	ids_tensor([self.batch_size]					,     self.type_sequence_label_size						)
									_snake_case     		=	ids_tensor([self.batch_size, self.seq_length]					,     self.num_labels						)
									_snake_case     		=	ids_tensor([self.batch_size]					,     self.num_choices						)

								_snake_case     		=	FunnelConfig(
								    vocab_size=self.vocab_size					,     block_sizes=self.block_sizes					,     num_decoder_layers=self.num_decoder_layers					,     d_model=self.d_model					,     n_head=self.n_head					,     d_head=self.d_head					,     d_inner=self.d_inner					,     hidden_act=self.hidden_act					,     hidden_dropout=self.hidden_dropout					,     attention_dropout=self.attention_dropout					,     activation_dropout=self.activation_dropout					,     max_position_embeddings=self.max_position_embeddings					,     type_vocab_size=self.type_vocab_size					,     initializer_std=self.initializer_std					,     )

								return (
								    config,
								    input_ids,
								    token_type_ids,
								    input_mask,
								    sequence_labels,
								    token_labels,
								    choice_labels,
								)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								_snake_case     		=	TFFunnelModel(config=lowerCAmelCase_						)
								_snake_case     		=	{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
								_snake_case     		=	model(lowerCAmelCase_						)

								_snake_case     		=	[input_ids, input_mask]
								_snake_case     		=	model(lowerCAmelCase_						)

								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.last_hidden_state.shape					,     (self.batch_size, self.seq_length, self.d_model)						)

								_snake_case     		=	False
								_snake_case     		=	TFFunnelModel(config=lowerCAmelCase_						)
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.last_hidden_state.shape					,     (self.batch_size, self.seq_length, self.d_model)						)

								_snake_case     		=	False
								_snake_case     		=	TFFunnelModel(config=lowerCAmelCase_						)
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.last_hidden_state.shape					,     (self.batch_size, self.seq_length, self.d_model)						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								_snake_case     		=	TFFunnelBaseModel(config=lowerCAmelCase_						)
								_snake_case     		=	{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
								_snake_case     		=	model(lowerCAmelCase_						)

								_snake_case     		=	[input_ids, input_mask]
								_snake_case     		=	model(lowerCAmelCase_						)

								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.last_hidden_state.shape					,     (self.batch_size, 2, self.d_model)						)

								_snake_case     		=	False
								_snake_case     		=	TFFunnelBaseModel(config=lowerCAmelCase_						)
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.last_hidden_state.shape					,     (self.batch_size, 3, self.d_model)						)

								_snake_case     		=	False
								_snake_case     		=	TFFunnelBaseModel(config=lowerCAmelCase_						)
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.last_hidden_state.shape					,     (self.batch_size, 2, self.d_model)						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								_snake_case     		=	TFFunnelForPreTraining(config=lowerCAmelCase_						)
								_snake_case     		=	{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.logits.shape					,     (self.batch_size, self.seq_length)						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								_snake_case     		=	TFFunnelForMaskedLM(config=lowerCAmelCase_						)
								_snake_case     		=	{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.logits.shape					,     (self.batch_size, self.seq_length, self.vocab_size)						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								_snake_case     		=	self.num_labels
								_snake_case     		=	TFFunnelForSequenceClassification(config=lowerCAmelCase_						)
								_snake_case     		=	{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.logits.shape					,     (self.batch_size, self.num_labels)						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								_snake_case     		=	self.num_choices
								_snake_case     		=	TFFunnelForMultipleChoice(config=lowerCAmelCase_						)
								_snake_case     		=	tf.tile(tf.expand_dims(lowerCAmelCase_					,     1						)					,     (1, self.num_choices, 1)						)
								_snake_case     		=	tf.tile(tf.expand_dims(lowerCAmelCase_					,     1						)					,     (1, self.num_choices, 1)						)
								_snake_case     		=	tf.tile(tf.expand_dims(lowerCAmelCase_					,     1						)					,     (1, self.num_choices, 1)						)
								_snake_case     		=	{
								    'input_ids': multiple_choice_inputs_ids,
								    'attention_mask': multiple_choice_input_mask,
								    'token_type_ids': multiple_choice_token_type_ids,
								}
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.logits.shape					,     (self.batch_size, self.num_choices)						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								_snake_case     		=	self.num_labels
								_snake_case     		=	TFFunnelForTokenClassification(config=lowerCAmelCase_						)
								_snake_case     		=	{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.logits.shape					,     (self.batch_size, self.seq_length, self.num_labels)						)


							def 				lowerCamelCase       ( self					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     lowerCAmelCase_					,     ):





								"""simple docstring"""






								_snake_case     		=	TFFunnelForQuestionAnswering(config=lowerCAmelCase_						)
								_snake_case     		=	{'input_ids': input_ids, 'attention_mask': input_mask, 'token_type_ids': token_type_ids}
								_snake_case     		=	model(lowerCAmelCase_						)
								self.parent.assertEqual(result.start_logits.shape					,     (self.batch_size, self.seq_length)						)
								self.parent.assertEqual(result.end_logits.shape					,     (self.batch_size, self.seq_length)						)




							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.prepare_config_and_inputs()
								(
								    (
								    _snake_case				
								)		,				(
								    _snake_case				
								)		,				(
								    _snake_case				
								)		,				(
								    _snake_case				
								)		,				(
								    _snake_case				
								)		,				(
								    _snake_case				
								)		,				(
								    _snake_case				
								)		,								
								)     		=	config_and_inputs
								_snake_case     		=	{'input_ids': input_ids, 'token_type_ids': token_type_ids, 'attention_mask': input_mask}
								return config, inputs_dict


@require_tf
class 		__UpperCAmelCase   (      _lowerCamelCase		,     _lowerCamelCase		,     unittest.TestCase		):
							__lowercase   	= (
							    (
							        TFFunnelModel,
							        TFFunnelForMaskedLM,
							        TFFunnelForPreTraining,
							        TFFunnelForQuestionAnswering,
							        TFFunnelForTokenClassification,
							    )
							    if is_tf_available()
							    else ()
							)
							__lowercase   	= (
							    {
							        """feature-extraction""": (TFFunnelBaseModel, TFFunnelModel),
							        """fill-mask""": TFFunnelForMaskedLM,
							        """question-answering""": TFFunnelForQuestionAnswering,
							        """text-classification""": TFFunnelForSequenceClassification,
							        """token-classification""": TFFunnelForTokenClassification,
							        """zero-shot""": TFFunnelForSequenceClassification,
							    }
							    if is_tf_available()
							    else {}
							)
							__lowercase   	= False
							__lowercase   	= False


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	TFFunnelModelTester(self						)
								_snake_case     		=	ConfigTester(self					,     config_class=lowerCAmelCase_						)


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								self.config_tester.run_common_tests()


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_model(*lowerCAmelCase_						)


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_pretraining(*lowerCAmelCase_						)


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_masked_lm(*lowerCAmelCase_						)


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_token_classification(*lowerCAmelCase_						)




							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_question_answering(*lowerCAmelCase_						)


@require_tf
class 		__UpperCAmelCase   (      _lowerCamelCase		,     unittest.TestCase		):
							__lowercase   	= (
							    (TFFunnelBaseModel, TFFunnelForMultipleChoice, TFFunnelForSequenceClassification) if is_tf_available() else ()
							)
							__lowercase   	= False
							__lowercase   	= False


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	TFFunnelModelTester(self					,     base=lowerCAmelCase_						)
								_snake_case     		=	ConfigTester(self					,     config_class=lowerCAmelCase_						)


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								self.config_tester.run_common_tests()


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_base_model(*lowerCAmelCase_						)


							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_sequence_classification(*lowerCAmelCase_						)




							def 				lowerCamelCase       ( self						):





								"""simple docstring"""






								_snake_case     		=	self.model_tester.prepare_config_and_inputs()
								self.model_tester.create_and_check_for_multiple_choice(*lowerCAmelCase_						)



'''simple docstring'''







import argparse
import re
from typing import Dict

import torch
from datasets import Audio, Dataset, load_dataset, load_metric

from transformers import AutoFeatureExtractor, pipeline







def        SCREAMING_SNAKE_CASE__			(     __A    ,       __A			) ->			int:
	_snake_case     		=	args.log_outputs
	_snake_case     		=	'_'.join(args.dataset.split('/'			) + [args.config, args.split]			)

	# load metric
	_snake_case     		=	load_metric('wer'			)
	_snake_case     		=	load_metric('cer'			)

	# compute metrics
	_snake_case     		=	wer.compute(references=result['target']    ,       predictions=result['prediction']			)
	_snake_case     		=	cer.compute(references=result['target']    ,       predictions=result['prediction']			)

	# print & log results
	_snake_case     		=	F'WER: {wer_result}\nCER: {cer_result}'
	print(__A			)

	with open(F'{dataset_id}_eval_results.txt'    ,       'w'			) as f:
		f.write(__A			)

	# log all results in text file. Possibly interesting for analysis
	if log_outputs is not None:
		_snake_case     		=	F'log_{dataset_id}_predictions.txt'
		_snake_case     		=	F'log_{dataset_id}_targets.txt'

		with open(__A    ,       'w'			) as p, open(__A    ,       'w'			) as t:
			# mapping function to write output
			def write_to_file(__A    ,       __A			):
				p.write(F'{i}' + '\n'			)
				p.write(batch['prediction'] + '\n'			)
				t.write(F'{i}' + '\n'			)
				t.write(batch['target'] + '\n'			)

			result.map(__A    ,       with_indices=__A			)







def        SCREAMING_SNAKE_CASE__			(     __A			) ->			str:
	_snake_case     		=	'[,?.!\-\;\:"“%‘”�—’…–]'  # noqa: W605 IMPORTANT: this should correspond to the chars that were ignored during training

	_snake_case     		=	re.sub(__A    ,       ''    ,       text.lower()			)

	# In addition, we can normalize the target text, e.g. removing new lines characters etc...
	# note that order is important here!
	_snake_case     		=	['\n\n', '\n', '   ', '  ']

	for t in token_sequences_to_ignore:
		_snake_case     		=	' '.join(text.split(__A			)			)

	return text







def        SCREAMING_SNAKE_CASE__			(     __A			) ->			Optional[int]:
	# load dataset
	_snake_case     		=	load_dataset(args.dataset    ,       args.config    ,       split=args.split    ,       use_auth_token=__A			)

	# for testing: only process the first two examples as a test
	# dataset = dataset.select(range(10))

	# load processor
	_snake_case     		=	AutoFeatureExtractor.from_pretrained(args.model_id			)
	_snake_case     		=	feature_extractor.sampling_rate

	# resample audio
	_snake_case     		=	dataset.cast_column('audio'    ,       Audio(sampling_rate=__A			)			)

	# load eval pipeline
	if args.device is None:
		_snake_case     		=	0 if torch.cuda.is_available() else -1
	_snake_case     		=	pipeline('automatic-speech-recognition'    ,       model=args.model_id    ,       device=args.device			)

	# map function to decode audio
	def map_to_pred(__A			):
		_snake_case     		=	asr(
		    batch['audio']['array']    ,       chunk_length_s=args.chunk_length_s    ,       stride_length_s=args.stride_length_s			)

		_snake_case     		=	prediction['text']
		_snake_case     		=	normalize_text(batch['sentence']			)
		return batch

	# run inference on all examples
	_snake_case     		=	dataset.map(__A    ,       remove_columns=dataset.column_names			)

	# compute and log_results
	# do not change function below
	log_results(__A    ,       __A			)


if __name__ == "__main__":
					lowercase					:							str   =      argparse.ArgumentParser()

					parser.add_argument(
					    "--model_id", type=str, required=True, help="Model identifier. Should be loadable with 🤗 Transformers"
					)
					parser.add_argument(
					    "--dataset",
					    type=str,
					    required=True,
					    help="Dataset name to evaluate the `model_id`. Should be loadable with 🤗 Datasets",
					)
					parser.add_argument(
					    "--config", type=str, required=True, help="Config of the dataset. *E.g.* `'en'`  for Common Voice"
					)
					parser.add_argument("--split", type=str, required=True, help="Split of the dataset. *E.g.* `'test'`")
					parser.add_argument(
					    "--chunk_length_s", type=float, default=None, help="Chunk length in seconds. Defaults to 5 seconds."
					)
					parser.add_argument(
					    "--stride_length_s", type=float, default=None, help="Stride of the audio chunks. Defaults to 1 second."
					)
					parser.add_argument(
					    "--log_outputs", action="store_true", help="If defined, write outputs to log file for analysis."
					)
					parser.add_argument(
					    "--device",
					    type=int,
					    default=None,
					    help="The device to run the pipeline on. -1 for CPU (default), 0 for the first GPU and so on.",
					)
					lowercase					:							Optional[int]   =      parser.parse_args()

					main(args)

