



"""simple docstring"""





from typing import List, Union

import numpy as np

from ..tokenization_utils import TruncationStrategy
from ..utils import add_end_docstrings, logging
from .base import PIPELINE_INIT_ARGS, ArgumentHandler, ChunkPipeline


lowercase					=       logging.get_logger(__name__)




class       lowercase__      (   _lowerCamelCase     ):
							'''simple docstring'''







							def 							lowerCamelCase_ (	self						,							snake_case    )      ->				Tuple:


									if isinstance(_SCREAMING_SNAKE_CASE						,							_SCREAMING_SNAKE_CASE    ):
											_UpperCAmelCase							=		[label.strip() for label in labels.split(','    ) if label.strip()]
									return labels






							def __call__(	self						,							snake_case						,							snake_case						,							snake_case    )      ->				Dict:


									if len(_SCREAMING_SNAKE_CASE    ) == 0 or len(_SCREAMING_SNAKE_CASE    ) == 0:
											raise ValueError('You must include at least one label and at least one sequence.'    )
									if hypothesis_template.format(labels[0]    ) == hypothesis_template:
											raise ValueError(
											    (
											        'The provided hypothesis_template "{}" was not able to be formatted with the target labels. '
											        'Make sure the passed template includes formatting syntax such as {{}} where the label should go.'
											    ).format(_SCREAMING_SNAKE_CASE    )    )

									if isinstance(_SCREAMING_SNAKE_CASE						,							_SCREAMING_SNAKE_CASE    ):
											_UpperCAmelCase							=		[sequences]

									_UpperCAmelCase							=		[]
									for sequence in sequences:
											sequence_pairs.extend([[sequence, hypothesis_template.format(_SCREAMING_SNAKE_CASE    )] for label in labels]    )

									return sequence_pairs, sequences







@add_end_docstrings(_lowerCamelCase     )
class       lowercase__      (   _lowerCamelCase     ):
							'''simple docstring'''







							def __init__(	self						,							snake_case=ZeroShotClassificationArgumentHandler()						,							*snake_case						,							**snake_case    )      ->				Union[str, Any]:
									_UpperCAmelCase							=		args_parser
									super().__init__(*_SCREAMING_SNAKE_CASE						,							**_SCREAMING_SNAKE_CASE    )
									if self.entailment_id == -1:
											logger.warning(
											    'Failed to determine \'entailment\' label id from the label2id mapping in the model config. Setting to '
											    '-1. Define a descriptive label2id mapping in the model config to ensure correct outputs.'    )






							@property
							def 							lowerCamelCase_ (	self    )      ->				Any:


									for label, ind in self.model.config.labelaid.items():
											if label.lower().startswith('entail'    ):
													return ind
									return -1






							def 							lowerCamelCase_ (	self						,							snake_case						,							snake_case=True						,							snake_case=True						,							snake_case=TruncationStrategy.ONLY_FIRST						,							**snake_case    )      ->				str:
									_UpperCAmelCase							=		self.framework
									if self.tokenizer.pad_token is None:
											# Override for tokenizers not supporting padding
											logger.error(
											    'Tokenizer was not supporting padding necessary for zero-shot, attempting to use '
											    ' `pad_token=eos_token`'    )
											_UpperCAmelCase							=		self.tokenizer.eos_token
									try:
											_UpperCAmelCase							=		self.tokenizer(
											    _SCREAMING_SNAKE_CASE						,							add_special_tokens=_SCREAMING_SNAKE_CASE						,							return_tensors=_SCREAMING_SNAKE_CASE						,							padding=_SCREAMING_SNAKE_CASE						,							truncation=_SCREAMING_SNAKE_CASE						,							)
									except Exception as e:
											if "too short" in str(_SCREAMING_SNAKE_CASE    ):
													# tokenizers might yell that we want to truncate
													# to a value that is not even reached by the input.
													# In that case we don't want to truncate.
													# It seems there's not a really better way to catch that
													# exception.

													_UpperCAmelCase							=		self.tokenizer(
													    _SCREAMING_SNAKE_CASE						,							add_special_tokens=_SCREAMING_SNAKE_CASE						,							return_tensors=_SCREAMING_SNAKE_CASE						,							padding=_SCREAMING_SNAKE_CASE						,							truncation=TruncationStrategy.DO_NOT_TRUNCATE						,							)
											else:
													raise e

									return inputs






							def 							lowerCamelCase_ (	self						,							**snake_case    )      ->				List[str]:


									if kwargs.get('multi_class'						,							_SCREAMING_SNAKE_CASE    ) is not None:
											_UpperCAmelCase							=		kwargs['multi_class']
											logger.warning(
											    'The `multi_class` argument has been deprecated and renamed to `multi_label`. '
											    '`multi_class` will be removed in a future version of Transformers.'    )
									_UpperCAmelCase							=		{}
									if "candidate_labels" in kwargs:
											_UpperCAmelCase							=		self._args_parser._parse_labels(kwargs['candidate_labels']    )
									if "hypothesis_template" in kwargs:
											_UpperCAmelCase							=		kwargs['hypothesis_template']

									_UpperCAmelCase							=		{}
									if "multi_label" in kwargs:
											_UpperCAmelCase							=		kwargs['multi_label']
									return preprocess_params, {}, postprocess_params






							def __call__(	self						,							snake_case						,							*snake_case						,							**snake_case						,							)      ->				List[Any]:


									if len(_SCREAMING_SNAKE_CASE    ) == 0:
											pass
									elif len(_SCREAMING_SNAKE_CASE    ) == 1 and "candidate_labels" not in kwargs:
											_UpperCAmelCase							=		args[0]
									else:
											raise ValueError(f'Unable to understand extra arguments {args}'    )

									return super().__call__(_SCREAMING_SNAKE_CASE						,							**_SCREAMING_SNAKE_CASE    )






							def 							lowerCamelCase_ (	self						,							snake_case						,							snake_case=None						,							snake_case="This example is {}."    )      ->				Any:
									_UpperCAmelCase							=		self._args_parser(_SCREAMING_SNAKE_CASE						,							_SCREAMING_SNAKE_CASE						,							_SCREAMING_SNAKE_CASE    )

									for i, (candidate_label, sequence_pair) in enumerate(zip(_SCREAMING_SNAKE_CASE						,							_SCREAMING_SNAKE_CASE    )    ):
											_UpperCAmelCase							=		self._parse_and_tokenize([sequence_pair]    )

											yield {
											    "candidate_label": candidate_label,
											    "sequence": sequences[0],
											    "is_last": i == len(_SCREAMING_SNAKE_CASE    ) - 1,
											    **model_input,
											}






							def 							lowerCamelCase_ (	self						,							snake_case    )      ->				Optional[int]:
									_UpperCAmelCase							=		inputs['candidate_label']
									_UpperCAmelCase							=		inputs['sequence']
									_UpperCAmelCase							=		{k: inputs[k] for k in self.tokenizer.model_input_names}
									_UpperCAmelCase							=		self.model(**_SCREAMING_SNAKE_CASE    )

									_UpperCAmelCase							=		{
									    'candidate_label': candidate_label,
									    'sequence': sequence,
									    'is_last': inputs['is_last'],
									    **outputs,
									}
									return model_outputs






							def 							lowerCamelCase_ (	self						,							snake_case						,							snake_case=False    )      ->				int:
									_UpperCAmelCase							=		[outputs['candidate_label'] for outputs in model_outputs]
									_UpperCAmelCase							=		[outputs['sequence'] for outputs in model_outputs]
									_UpperCAmelCase							=		np.concatenate([output['logits'].numpy() for output in model_outputs]    )
									_UpperCAmelCase							=		logits.shape[0]
									_UpperCAmelCase							=		len(_SCREAMING_SNAKE_CASE    )
									_UpperCAmelCase							=		N // n
									_UpperCAmelCase							=		logits.reshape((num_sequences, n, -1)    )

									if multi_label or len(_SCREAMING_SNAKE_CASE    ) == 1:
											# softmax over the entailment vs. contradiction dim for each label independently
											_UpperCAmelCase							=		self.entailment_id
											_UpperCAmelCase							=		-1 if entailment_id == 0 else 0
											_UpperCAmelCase							=		reshaped_outputs[..., [contradiction_id, entailment_id]]
											_UpperCAmelCase							=		np.exp(_SCREAMING_SNAKE_CASE    ) / np.exp(_SCREAMING_SNAKE_CASE    ).sum(-1						,							keepdims=_SCREAMING_SNAKE_CASE    )
											_UpperCAmelCase							=		scores[..., 1]
									else:
											# softmax the "entailment" logits over all candidate labels
											_UpperCAmelCase							=		reshaped_outputs[..., self.entailment_id]
											_UpperCAmelCase							=		np.exp(_SCREAMING_SNAKE_CASE    ) / np.exp(_SCREAMING_SNAKE_CASE    ).sum(-1						,							keepdims=_SCREAMING_SNAKE_CASE    )

									_UpperCAmelCase							=		list(reversed(scores[0].argsort()    )    )
									return {
									    "sequence": sequences[0],
									    "labels": [candidate_labels[i] for i in top_inds],
									    "scores": scores[0, top_inds].tolist(),
									}







"""simple docstring"""

import argparse
import json

from tqdm import tqdm
def  __snake_case							(	)     ->       Any:






				'''simple docstring'''
				SCREAMING_SNAKE_CASE						:							str							=     argparse.ArgumentParser()

				# Required parameters
				parser.add_argument(
				    '--src_path'							,		type=__A							,		default='biencoder-nq-dev.json'							,		help='Path to raw DPR training data'							,		)
				parser.add_argument(
				    '--evaluation_set'							,		type=__A							,		help='where to store parsed evaluation_set file'							,		)
				parser.add_argument(
				    '--gold_data_path'							,		type=__A							,		help='where to store parsed gold_data_path file'							,		)
				SCREAMING_SNAKE_CASE						:							Tuple							=     parser.parse_args()

				with open(args.src_path							,		'r'     ) as src_file, open(args.evaluation_set							,		'w'     ) as eval_file, open(
				    args.gold_data_path							,		'w'     ) as gold_file:
								SCREAMING_SNAKE_CASE						:							Tuple							=     json.load(__A     )
								for dpr_record in tqdm(__A     ):
												SCREAMING_SNAKE_CASE						:							Tuple							=     dpr_record['question']
												SCREAMING_SNAKE_CASE						:							Optional[int]							=     [context['title'] for context in dpr_record['positive_ctxs']]
												eval_file.write(question + '\n'     )
												gold_file.write('\t'.join(__A     ) + '\n'     )


if __name__ == "__main__":
							main()





