"""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    __snake_case       (     snake_case__				):



      def   lowerCamelCase_   (						self    ,			lowercase)	->			Union[str, Any]:





        '''simple docstring'''





        if isinstance(_A    ,			_A):
          a__:     Optional[Any]										=	[label.strip() for label in labels.split(',') if label.strip()]
        return labels



      def __call__(						self    ,			lowercase    ,			lowercase    ,			lowercase)	->			Optional[int]:





        '''simple docstring'''





        if len(_A) == 0 or len(_A) == 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(_A))

        if isinstance(_A    ,			_A):
          a__:     Optional[int]										=	[sequences]

        a__:     Any										=	[]
        for sequence in sequences:
          sequence_pairs.extend([[sequence, hypothesis_template.format(_A)] for label in labels])

        return sequence_pairs, sequences







@add_end_docstrings(snake_case__				)
class    __snake_case       (     snake_case__				):



      def __init__(						self    ,			lowercase=ZeroShotClassificationArgumentHandler()    ,			*lowercase    ,			**lowercase)	->			List[Any]:





        '''simple docstring'''





        a__:     Optional[int]										=	args_parser
        super().__init__(*_A    ,			**_A)
        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)	->			str:





        '''simple docstring'''





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



      def   lowerCamelCase_   (						self    ,			lowercase    ,			lowercase=True    ,			lowercase=True    ,			lowercase=TruncationStrategy.ONLY_FIRST    ,			**lowercase)	->			Any:





        '''simple docstring'''





        a__:     Union[str, Any]										=	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`')
          a__:     Tuple										=	self.tokenizer.eos_token
        try:
          a__:     Tuple										=	self.tokenizer(
              _A    ,			add_special_tokens=_A    ,			return_tensors=_A    ,			padding=_A    ,			truncation=_A    ,			)
        except Exception as e:
          if "too short" in str(_A):
            # 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.

            a__:     Optional[int]										=	self.tokenizer(
                _A    ,			add_special_tokens=_A    ,			return_tensors=_A    ,			padding=_A    ,			truncation=TruncationStrategy.DO_NOT_TRUNCATE    ,			)
          else:
            raise e

        return inputs



      def   lowerCamelCase_   (						self    ,			**lowercase)	->			Optional[int]:





        '''simple docstring'''





        if kwargs.get('multi_class'    ,			_A) is not None:
          a__:     Optional[int]										=	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.')
        a__:     int										=	{}
        if "candidate_labels" in kwargs:
          a__:     Union[str, Any]										=	self._args_parser._parse_labels(kwargs['candidate_labels'])
        if "hypothesis_template" in kwargs:
          a__:     int										=	kwargs['hypothesis_template']

        a__:     Dict										=	{}
        if "multi_label" in kwargs:
          a__:     Union[str, Any]										=	kwargs['multi_label']
        return preprocess_params, {}, postprocess_params



      def __call__(						self    ,			lowercase    ,			*lowercase    ,			**lowercase    ,			)	->			Union[str, Any]:





        '''simple docstring'''





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

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



      def   lowerCamelCase_   (						self    ,			lowercase    ,			lowercase=None    ,			lowercase="This example is {}.")	->			Tuple:





        '''simple docstring'''





        a__:     Union[str, Any]										=	self._args_parser(_A    ,			_A    ,			_A)

        for i, (candidate_label, sequence_pair) in enumerate(zip(_A    ,			_A)):
          a__:     Union[str, Any]										=	self._parse_and_tokenize([sequence_pair])

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



      def   lowerCamelCase_   (						self    ,			lowercase)	->			Union[str, Any]:





        '''simple docstring'''





        a__:     Dict										=	inputs['candidate_label']
        a__:     Optional[Any]										=	inputs['sequence']
        a__:     List[Any]										=	{k: inputs[k] for k in self.tokenizer.model_input_names}
        a__:     Optional[Any]										=	self.model(**_A)

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



      def   lowerCamelCase_   (						self    ,			lowercase    ,			lowercase=False)	->			Tuple:





        '''simple docstring'''





        a__:     List[Any]										=	[outputs['candidate_label'] for outputs in model_outputs]
        a__:     Optional[Any]										=	[outputs['sequence'] for outputs in model_outputs]
        a__:     Optional[Any]										=	np.concatenate([output['logits'].numpy() for output in model_outputs])
        a__:     str										=	logits.shape[0]
        a__:     Optional[int]										=	len(_A)
        a__:     Tuple										=	N // n
        a__:     Optional[int]										=	logits.reshape((num_sequences, n, -1))

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

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









import argparse
import json

from tqdm import tqdm







def  _SCREAMING_SNAKE_CASE				(   )      ->      List[Any]:
					__A	:						Tuple       =				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'					,						)
					__A	:						Optional[int]       =				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:
										__A	:						List[Any]       =				json.load(a  )
										for dpr_record in tqdm(a  ):
															__A	:						Dict       =				dpr_record['question']
															__A	:						Any       =				[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()




