



"""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


A		=						logging.get_logger(__name__)

class       _a  ( SCREAMING_SNAKE_CASE__):



			def 				__lowercase					(			self							:				int   ,	_lowercase							:				int		)							->					List[str]:
									if isinstance(_lowercase   ,	_lowercase		):
															snake_case   :			List[str]						    = [label.strip() for label in labels.split(","		) if label.strip()]
									return labels



			def __call__(			self							:				Optional[Any]   ,	_lowercase							:				Dict   ,	_lowercase							:				Union[str, Any]   ,	_lowercase							:				List[Any]		)							->					Optional[int]:
									if len(_lowercase		) == 0 or len(_lowercase		) == 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(_lowercase		)		)

									if isinstance(_lowercase   ,	_lowercase		):
															snake_case   :			Optional[Any]						    = [sequences]

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

									return sequence_pairs, sequences



@add_end_docstrings(SCREAMING_SNAKE_CASE__)
class       _a  ( SCREAMING_SNAKE_CASE__):



			def __init__(			self							:				List[str]   ,	_lowercase							:				Union[str, Any]=ZeroShotClassificationArgumentHandler()   ,	*_lowercase							:				List[Any]   ,	**_lowercase							:				Union[str, Any]		)							->					List[Any]:
									snake_case   :			Optional[Any]						    = args_parser
									super().__init__(*_lowercase   ,	**_lowercase		)
									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 				__lowercase					(			self							:				str		)							->					Dict:
									for label, ind in self.model.config.labelaid.items():
															if label.lower().startswith("entail"		):
																					return ind
									return -1



			def 				__lowercase					(			self							:				Optional[int]   ,	_lowercase							:				List[Any]   ,	_lowercase							:				Optional[Any]=True   ,	_lowercase							:				int=True   ,	_lowercase							:				Any=TruncationStrategy.ONLY_FIRST   ,	**_lowercase							:				Optional[int]		)							->					str:
									snake_case   :			Optional[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`"		)
															snake_case   :			Optional[int]						    = self.tokenizer.eos_token
									try:
															snake_case   :			List[str]						    = self.tokenizer(
															    _lowercase   ,	add_special_tokens=_lowercase   ,	return_tensors=_lowercase   ,	padding=_lowercase   ,	truncation=_lowercase   ,	)
									except Exception as e:
															if "too short" in str(_lowercase		):
																					# 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.

																					snake_case   :			List[str]						    = self.tokenizer(
																					    _lowercase   ,	add_special_tokens=_lowercase   ,	return_tensors=_lowercase   ,	padding=_lowercase   ,	truncation=TruncationStrategy.DO_NOT_TRUNCATE   ,	)
															else:
																					raise e

									return inputs



			def 				__lowercase					(			self							:				Tuple   ,	**_lowercase							:				List[Any]		)							->					Optional[Any]:
									if kwargs.get("multi_class"   ,	_lowercase		) is not None:
															snake_case   :			Union[str, Any]						    = 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."		)
									snake_case   :			List[str]						    = {}
									if "candidate_labels" in kwargs:
															snake_case   :			int						    = self._args_parser._parse_labels(kwargs["candidate_labels"]		)
									if "hypothesis_template" in kwargs:
															snake_case   :			List[Any]						    = kwargs["hypothesis_template"]

									snake_case   :			Any						    = {}
									if "multi_label" in kwargs:
															snake_case   :			Tuple						    = kwargs["multi_label"]
									return preprocess_params, {}, postprocess_params



			def __call__(			self							:				Optional[int]   ,	_lowercase							:				Union[str, List[str]]   ,	*_lowercase							:				Dict   ,	**_lowercase							:				Optional[int]   ,	)							->					List[str]:
									if len(_lowercase		) == 0:
															pass
									elif len(_lowercase		) == 1 and "candidate_labels" not in kwargs:
															snake_case   :			List[Any]						    = args[0]
									else:
															raise ValueError(F'''Unable to understand extra arguments {args}'''		)

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



			def 				__lowercase					(			self							:				Optional[int]   ,	_lowercase							:				List[str]   ,	_lowercase							:				Dict=None   ,	_lowercase							:				Union[str, Any]="This example is {}."		)							->					Optional[int]:
									snake_case    ,			snake_case   :			List[str]						    = self._args_parser(_lowercase   ,	_lowercase   ,	_lowercase		)

									for i, (candidate_label, sequence_pair) in enumerate(zip(_lowercase   ,	_lowercase		)		):
															snake_case   :			Tuple						    = self._parse_and_tokenize([sequence_pair]		)

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



			def 				__lowercase					(			self							:				Union[str, Any]   ,	_lowercase							:				Union[str, Any]		)							->					Any:
									snake_case   :			int						    = inputs["candidate_label"]
									snake_case   :			Dict						    = inputs["sequence"]
									snake_case   :			Optional[int]						    = {k: inputs[k] for k in self.tokenizer.model_input_names}
									snake_case   :			str						    = self.model(**_lowercase		)

									snake_case   :			Optional[int]						    = {
									    "candidate_label": candidate_label,
									    "sequence": sequence,
									    "is_last": inputs["is_last"],
									    **outputs,
									}
									return model_outputs



			def 				__lowercase					(			self							:				int   ,	_lowercase							:				Union[str, Any]   ,	_lowercase							:				Dict=False		)							->					Dict:
									snake_case   :			int						    = [outputs["candidate_label"] for outputs in model_outputs]
									snake_case   :			Tuple						    = [outputs["sequence"] for outputs in model_outputs]
									snake_case   :			int						    = np.concatenate([output["logits"].numpy() for output in model_outputs]		)
									snake_case   :			int						    = logits.shape[0]
									snake_case   :			Any						    = len(_lowercase		)
									snake_case   :			Optional[int]						    = N // n
									snake_case   :			List[Any]						    = logits.reshape((num_sequences, n, -1)		)

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

									snake_case   :			List[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(),
									}








"""simple docstring"""







import os
import textwrap

import pyarrow as pa
import pytest

from datasets import ClassLabel, Features, Image
from datasets.packaged_modules.csv.csv import Csv

from ..utils import require_pil

@pytest.fixture
def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							Optional[Any] ):
						"""simple docstring"""






						snake_case   :			str						    = tmp_path / "file.csv"
						snake_case   :			Tuple						    = textwrap.dedent(
						    "\\n        header1,header2\n        1,2\n        10,20\n        " )
						with open(lowerCamelCase_     ,		"w" ) as f:
												f.write(lowerCamelCase_ )
						return str(lowerCamelCase_ )

@pytest.fixture
def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							List[str] ):
						"""simple docstring"""






						snake_case   :			List[Any]						    = tmp_path / "malformed_file.csv"
						snake_case   :			Union[str, Any]						    = textwrap.dedent(
						    "\\n        header1,header2\n        1,2\n        10,20,\n        " )
						with open(lowerCamelCase_     ,		"w" ) as f:
												f.write(lowerCamelCase_ )
						return str(lowerCamelCase_ )

@pytest.fixture
def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							int     ,		lowerCamelCase_:							Optional[Any] ):
						"""simple docstring"""






						snake_case   :			Dict						    = tmp_path / "csv_with_image.csv"
						snake_case   :			List[Any]						    = textwrap.dedent(
						    f'''\
        image
        {image_file}
        ''' )
						with open(lowerCamelCase_     ,		"w" ) as f:
												f.write(lowerCamelCase_ )
						return str(lowerCamelCase_ )

@pytest.fixture
def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							List[str] ):
						"""simple docstring"""






						snake_case   :			Optional[Any]						    = tmp_path / "csv_with_label.csv"
						snake_case   :			Tuple						    = textwrap.dedent(
						    "\\n        label\n        good\n        bad\n        good\n        " )
						with open(lowerCamelCase_     ,		"w" ) as f:
												f.write(lowerCamelCase_ )
						return str(lowerCamelCase_ )

@pytest.fixture
def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							List[Any] ):
						"""simple docstring"""






						snake_case   :			Union[str, Any]						    = tmp_path / "csv_with_int_list.csv"
						snake_case   :			Optional[int]						    = textwrap.dedent(
						    "\\n        int_list\n        1 2 3\n        4 5 6\n        7 8 9\n        " )
						with open(lowerCamelCase_     ,		"w" ) as f:
												f.write(lowerCamelCase_ )
						return str(lowerCamelCase_ )

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							List[str]     ,		lowerCamelCase_:							Any     ,		lowerCamelCase_:							List[str] ):
						"""simple docstring"""






						snake_case   :			Tuple						    = Csv()
						snake_case   :			str						    = csv._generate_tables([[csv_file, malformed_csv_file]] )
						with pytest.raises(lowerCamelCase_     ,		match="Error tokenizing data" ):
												for _ in generator:
																		pass
						assert any(
						    record.levelname == "ERROR"
						    and "Failed to read file" in record.message
						    and os.path.basename(lowerCamelCase_ ) in record.message
						    for record in caplog.records )

@require_pil
def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							str ):
						"""simple docstring"""






						with open(lowerCamelCase_     ,		encoding="utf-8" ) as f:
												snake_case   :			List[Any]						    = f.read().splitlines()[1]
						snake_case   :			List[str]						    = Csv(encoding="utf-8"     ,		features=Features({"image": Image()} ) )
						snake_case   :			Optional[int]						    = csv._generate_tables([[csv_file_with_image]] )
						snake_case   :			List[Any]						    = pa.concat_tables([table for _, table in generator] )
						assert pa_table.schema.field("image" ).type == Image()()
						snake_case   :			str						    = pa_table.to_pydict()["image"]
						assert generated_content == [{"path": image_file, "bytes": None}]

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							int ):
						"""simple docstring"""






						with open(lowerCamelCase_     ,		encoding="utf-8" ) as f:
												snake_case   :			Optional[Any]						    = f.read().splitlines()[1:]
						snake_case   :			List[str]						    = Csv(encoding="utf-8"     ,		features=Features({"label": ClassLabel(names=["good", "bad"] )} ) )
						snake_case   :			int						    = csv._generate_tables([[csv_file_with_label]] )
						snake_case   :			Dict						    = pa.concat_tables([table for _, table in generator] )
						assert pa_table.schema.field("label" ).type == ClassLabel(names=["good", "bad"] )()
						snake_case   :			Dict						    = pa_table.to_pydict()["label"]
						assert generated_content == [ClassLabel(names=["good", "bad"] ).straint(lowerCamelCase_ ) for label in labels]

def  __SCREAMING_SNAKE_CASE		(			lowerCamelCase_:							Union[str, Any] ):
						"""simple docstring"""






						snake_case   :			List[Any]						    = Csv(encoding="utf-8"     ,		sep=","     ,		converters={"int_list": lambda lowerCamelCase_ : [int(lowerCamelCase_ ) for i in x.split()]} )
						snake_case   :			Tuple						    = csv._generate_tables([[csv_file_with_int_list]] )
						snake_case   :			str						    = pa.concat_tables([table for _, table in generator] )
						assert pa.types.is_list(pa_table.schema.field("int_list" ).type )
						snake_case   :			Any						    = pa_table.to_pydict()["int_list"]
						assert generated_content == [[1, 2, 3], [4, 5, 6], [7, 8, 9]]



