





'''simple docstring'''






import subprocess
from typing import Union

import numpy as np
import requests

from ..utils import add_end_docstrings, is_torch_available, is_torchaudio_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_torch_available():
	from ..models.auto.modeling_auto import MODEL_FOR_AUDIO_CLASSIFICATION_MAPPING

__UpperCamelCase      :   Optional[int]								=     logging.get_logger(__name__)





def      lowercase  (  lowerCAmelCase :						bytes   , lowerCAmelCase :						int):



				"""simple docstring"""


				_A :				Any				=  f"""{sampling_rate}"""
				_A :				Optional[int]				=  '''1'''
				_A :				Dict				=  '''f32le'''
				_A :				List[Any]				=  [
				    '''ffmpeg''',
				    '''-i''',
				    '''pipe:0''',
				    '''-ac''',
				    ac,
				    '''-ar''',
				    ar,
				    '''-f''',
				    format_for_conversion,
				    '''-hide_banner''',
				    '''-loglevel''',
				    '''quiet''',
				    '''pipe:1''',
				]

				try:
								_A :				Optional[int]				=  subprocess.Popen(lowerCAmelCase   , stdin=subprocess.PIPE   , stdout=subprocess.PIPE)
				except FileNotFoundError:
								raise ValueError('''ffmpeg was not found but is required to load audio files from filename''')
				_A :				List[Any]				=  ffmpeg_process.communicate(lowerCAmelCase)
				_A :				Optional[Any]				=  output_stream[0]

				_A :				Optional[int]				=  np.frombuffer(lowerCAmelCase   , np.floataa)
				if audio.shape[0] == 0:
								raise ValueError('''Malformed soundfile''')
				return audio




@add_end_docstrings(snake_case_						)
class  lowerCamelCase__				(						snake_case_						):


	"""simple docstring"""





	def __init__(					self   ,     *UpperCAmelCase__   ,     **UpperCAmelCase__			)       ->				Dict:
					# Default, might be overriden by the model.config.
					_A :				Any				=  5
					super().__init__(*UpperCAmelCase__   ,     **UpperCAmelCase__			)

					if self.framework != "pt":
									raise ValueError(F"""The {self.__class__} is only available in PyTorch."""			)

					self.check_model_type(UpperCAmelCase__			)




	def __call__(					self   ,     UpperCAmelCase__   ,     **UpperCAmelCase__   ,     )       ->				Dict:
					return super().__call__(UpperCAmelCase__   ,     **UpperCAmelCase__			)




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__=None   ,     **UpperCAmelCase__			)       ->				int:
					# No parameters on this pipeline right now
					_A :				Optional[Any]				=  {}
					if top_k is not None:
									if top_k > self.model.config.num_labels:
													_A :				List[Any]				=  self.model.config.num_labels
									_A :				Optional[int]				=  top_k
					return {}, {}, postprocess_params




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__			)       ->				Union[str, Any]:
					if isinstance(UpperCAmelCase__   ,     UpperCAmelCase__			):
									if inputs.startswith('''http://'''			) or inputs.startswith('''https://'''			):
													# We need to actually check for a real protocol, otherwise it's impossible to use a local file
													# like http_huggingface_co.png
													_A :				Tuple				=  requests.get(UpperCAmelCase__			).content
									else:
													with open(UpperCAmelCase__   ,     '''rb'''			) as f:
																	_A :				Dict				=  f.read()

					if isinstance(UpperCAmelCase__   ,     UpperCAmelCase__			):
									_A :				Optional[int]				=  ffmpeg_read(UpperCAmelCase__   ,     self.feature_extractor.sampling_rate			)

					if isinstance(UpperCAmelCase__   ,     UpperCAmelCase__			):
									# Accepting `"array"` which is the key defined in `datasets` for
									# better integration
									if not ("sampling_rate" in inputs and ("raw" in inputs or "array" in inputs)):
													raise ValueError(
													    '''When passing a dictionary to AudioClassificationPipeline, the dict needs to contain a '''
													    '''"raw" key containing the numpy array representing the audio and a "sampling_rate" key, '''
													    '''containing the sampling_rate associated with that array'''			)

									_A :				List[str]				=  inputs.pop('''raw'''   ,     UpperCAmelCase__			)
									if _inputs is None:
													# Remove path which will not be used from `datasets`.
													inputs.pop('''path'''   ,     UpperCAmelCase__			)
													_A :				Optional[Any]				=  inputs.pop('''array'''   ,     UpperCAmelCase__			)
									_A :				List[str]				=  inputs.pop('''sampling_rate'''			)
									_A :				Optional[Any]				=  _inputs
									if in_sampling_rate != self.feature_extractor.sampling_rate:
													import torch

													if is_torchaudio_available():
																	from torchaudio import functional as F
													else:
																	raise ImportError(
																	    '''torchaudio is required to resample audio samples in AudioClassificationPipeline. '''
																	    '''The torchaudio package can be installed through: `pip install torchaudio`.'''			)

													_A :				Union[str, Any]				=  F.resample(
													    torch.from_numpy(UpperCAmelCase__			)   ,     UpperCAmelCase__   ,     self.feature_extractor.sampling_rate			).numpy()

					if not isinstance(UpperCAmelCase__   ,     np.ndarray			):
									raise ValueError('''We expect a numpy ndarray as input'''			)
					if len(inputs.shape			) != 1:
									raise ValueError('''We expect a single channel audio input for AudioClassificationPipeline'''			)

					_A :				Any				=  self.feature_extractor(
					    UpperCAmelCase__   ,     sampling_rate=self.feature_extractor.sampling_rate   ,     return_tensors='''pt'''			)
					return processed




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__			)       ->				Tuple:
					_A :				List[Any]				=  self.model(**UpperCAmelCase__			)
					return model_outputs




	def 				_lowerCamelCase				(					self   ,     UpperCAmelCase__   ,     UpperCAmelCase__=5			)       ->				Optional[Any]:
					_A :				Optional[Any]				=  model_outputs.logits[0].softmax(-1			)
					_A	,       _A :				Any				=  probs.topk(UpperCAmelCase__			)

					_A :				Optional[Any]				=  scores.tolist()
					_A :				int				=  ids.tolist()

					_A :				int				=  [{'''score''': score, '''label''': self.model.config.idalabel[_id]} for score, _id in zip(UpperCAmelCase__   ,     UpperCAmelCase__			)]

					return labels







'''simple docstring'''






import argparse
import json
import re
import string
import sys
from collections import Counter





def      lowercase  (  lowerCAmelCase :						int):



				"""simple docstring"""


				def remove_articles(lowerCAmelCase :						List[Any]):
								return re.sub(R'''\b(a|an|the)\b'''   , ''' '''   , lowerCAmelCase)

				def white_space_fix(lowerCAmelCase :						Tuple):
								return " ".join(text.split())

				def remove_punc(lowerCAmelCase :						str):
								_A :				List[Any]				=  set(string.punctuation)
								return "".join(ch for ch in text if ch not in exclude)

				def lower(lowerCAmelCase :						Any):
								return text.lower()

				return white_space_fix(remove_articles(remove_punc(lower(lowerCAmelCase))))





def      lowercase  (  lowerCAmelCase :						int   , lowerCAmelCase :						Dict):



				"""simple docstring"""


				_A :				str				=  normalize_answer(lowerCAmelCase).split()
				_A :				Dict				=  normalize_answer(lowerCAmelCase).split()
				_A :				str				=  Counter(lowerCAmelCase) & Counter(lowerCAmelCase)
				_A :				List[Any]				=  sum(common.values())
				if num_same == 0:
								return 0
				_A :				Tuple				=  1.0 * num_same / len(lowerCAmelCase)
				_A :				List[Any]				=  1.0 * num_same / len(lowerCAmelCase)
				_A :				int				=  (2 * precision * recall) / (precision + recall)
				return fa





def      lowercase  (  lowerCAmelCase :						Any   , lowerCAmelCase :						str):



				"""simple docstring"""


				return normalize_answer(lowerCAmelCase) == normalize_answer(lowerCAmelCase)





def      lowercase  (  lowerCAmelCase :						Optional[Any]   , lowerCAmelCase :						Optional[int]   , lowerCAmelCase :						Optional[int]):



				"""simple docstring"""


				_A :				Optional[int]				=  []
				for ground_truth in ground_truths:
								_A :				Union[str, Any]				=  metric_fn(lowerCAmelCase   , lowerCAmelCase)
								scores_for_ground_truths.append(lowerCAmelCase)
				return max(lowerCAmelCase)





def      lowercase  (  lowerCAmelCase :						int   , lowerCAmelCase :						str):



				"""simple docstring"""


				_A :				Union[str, Any]				=  0
				_A :				List[Any]				=  []
				for passage in dataset:
								for qa in passage["qas"]:
												total += 1
												if qa["id"] not in predictions:
																_A :				str				=  f"""Unanswered question {qa['id']} will receive score 0."""
																print(lowerCAmelCase   , file=sys.stderr)
																continue

												_A :				Optional[int]				=  [x['''text'''] for x in qa['''answers''']]
												_A :				Union[str, Any]				=  predictions[qa['''id''']]

												_A :				Optional[Any]				=  metric_max_over_ground_truths(lowerCAmelCase   , lowerCAmelCase   , lowerCAmelCase)
												if int(_exact_match) == 1:
																correct_ids.append(qa['''id'''])
												exact_match += _exact_match

												fa += metric_max_over_ground_truths(lowerCAmelCase   , lowerCAmelCase   , lowerCAmelCase)

				_A :				Dict				=  exact_match / total
				_A :				Dict				=  fa / total

				return {"exact_match": exact_match, "f1": fa}, correct_ids


if __name__ == "__main__":
	__UpperCamelCase      :   List[str]								=     '''1.0'''
	__UpperCamelCase      :   Union[str, Any]								=     argparse.ArgumentParser('''Official evaluation script for ReCoRD v1.0.''')
	parser.add_argument('''data_file''', help='''The dataset file in JSON format.''')
	parser.add_argument('''pred_file''', help='''The model prediction file in JSON format.''')
	parser.add_argument('''--output_correct_ids''', action='''store_true''', help='''Output the correctly answered query IDs.''')
	__UpperCamelCase      :   Optional[int]								=     parser.parse_args()

	with open(args.data_file) as data_file:
		__UpperCamelCase      :   Optional[int]								=     json.load(data_file)
		if dataset_json["version"] != expected_version:
			print(
			    f'Evaluation expects v-{expected_version}, but got dataset with v-{dataset_json["version"]}',
			    file=sys.stderr,
			)
		__UpperCamelCase      :   Optional[Any]								=     dataset_json['''data''']

	with open(args.pred_file) as pred_file:
		__UpperCamelCase      :   int								=     json.load(pred_file)

	__UpperCamelCase							,  __UpperCamelCase      :   List[str]								=     evaluate(dataset, predictions)

	if args.output_correct_ids:
		print(f'Output {len(correct_ids)} correctly answered question IDs.')
		with open('''correct_ids.json''', '''w''') as f:
			json.dump(correct_ids, f)
