




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


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

class        _a    (						__lowerCAmelCase							):






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






     def __call__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE ,_SCREAMING_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   ):
               _snake_case							=				[sequences]

          _snake_case							=				[]
          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        _a    (						__lowerCAmelCase							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE=ZeroShotClassificationArgumentHandler() ,*_SCREAMING_SNAKE_CASE ,**_SCREAMING_SNAKE_CASE   )							->      Optional[Any]:
          _snake_case							=				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     _lowercase     (			self   )							->      Optional[Any]:
          for label, ind in self.model.config.labelaid.items():
               if label.lower().startswith("entail"   ):
                    return ind
          return -1






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE=True ,_SCREAMING_SNAKE_CASE=True ,_SCREAMING_SNAKE_CASE=TruncationStrategy.ONLY_FIRST ,**_SCREAMING_SNAKE_CASE   )							->      Union[str, Any]:
          _snake_case							=				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							=				self.tokenizer.eos_token
          try:
               _snake_case							=				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.

                    _snake_case							=				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     _lowercase     (			self ,**_SCREAMING_SNAKE_CASE   )							->      List[Any]:
          if kwargs.get("multi_class" ,_SCREAMING_SNAKE_CASE   ) is not None:
               _snake_case							=				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							=				{}
          if "candidate_labels" in kwargs:
               _snake_case							=				self._args_parser._parse_labels(kwargs["candidate_labels"]   )
          if "hypothesis_template" in kwargs:
               _snake_case							=				kwargs["hypothesis_template"]

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






     def __call__(			self ,_SCREAMING_SNAKE_CASE ,*_SCREAMING_SNAKE_CASE ,**_SCREAMING_SNAKE_CASE ,)							->      List[str]:
          if len(_SCREAMING_SNAKE_CASE   ) == 0:
               pass
          elif len(_SCREAMING_SNAKE_CASE   ) == 1 and "candidate_labels" not in kwargs:
               _snake_case							=				args[0]
          else:
               raise ValueError(f"""Unable to understand extra arguments {args}"""   )

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






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE=None ,_SCREAMING_SNAKE_CASE="This example is {}."   )							->      int:
          _snake_case							, _snake_case							=				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   )   ):
               _snake_case							=				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     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      Tuple:
          _snake_case							=				inputs["candidate_label"]
          _snake_case							=				inputs["sequence"]
          _snake_case							=				{k: inputs[k] for k in self.tokenizer.model_input_names}
          _snake_case							=				self.model(**_SCREAMING_SNAKE_CASE   )

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






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE=False   )							->      Union[str, Any]:
          _snake_case							=				[outputs["candidate_label"] for outputs in model_outputs]
          _snake_case							=				[outputs["sequence"] for outputs in model_outputs]
          _snake_case							=				np.concatenate([output["logits"].numpy() for output in model_outputs]   )
          _snake_case							=				logits.shape[0]
          _snake_case							=				len(_SCREAMING_SNAKE_CASE   )
          _snake_case							=				N // n
          _snake_case							=				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
               _snake_case							=				self.entailment_id
               _snake_case							=				-1 if entailment_id == 0 else 0
               _snake_case							=				reshaped_outputs[..., [contradiction_id, entailment_id]]
               _snake_case							=				np.exp(_SCREAMING_SNAKE_CASE   ) / np.exp(_SCREAMING_SNAKE_CASE   ).sum(-1 ,keepdims=_SCREAMING_SNAKE_CASE   )
               _snake_case							=				scores[..., 1]
          else:
               # softmax the "entailment" logits over all candidate labels
               _snake_case							=				reshaped_outputs[..., self.entailment_id]
               _snake_case							=				np.exp(_SCREAMING_SNAKE_CASE   ) / np.exp(_SCREAMING_SNAKE_CASE   ).sum(-1 ,keepdims=_SCREAMING_SNAKE_CASE   )

          _snake_case							=				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'''





from __future__ import annotations

from collections.abc import Callable
from typing import Generic, TypeVar

UpperCamelCase_						:      int       =						TypeVar('''T''')
UpperCamelCase_						:      Dict       =						TypeVar('''U''')

class        _a    (						Generic[T, U]							):






     def __init__(			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      Dict:
          _snake_case							=				key
          _snake_case							=				val
          _snake_case							=				None
          _snake_case							=				None






     def __repr__(			self   )							->      str:
          return (
              f"""Node: key: {self.key}, val: {self.val}, """
              f"""has next: {bool(self.next   )}, has prev: {bool(self.prev   )}"""
          )







class        _a    (						Generic[T, U]							):






     def __init__(			self   )							->      None:
          _snake_case							=				DoubleLinkedListNode(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          _snake_case							=				DoubleLinkedListNode(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
          _snake_case							, _snake_case							=				self.rear, self.head






     def __repr__(			self   )							->      str:
          _snake_case							=				["DoubleLinkedList"]
          _snake_case							=				self.head
          while node.next is not None:
               rep.append(str(_SCREAMING_SNAKE_CASE   )   )
               _snake_case							=				node.next
          rep.append(str(self.rear   )   )
          return ",\n    ".join(_SCREAMING_SNAKE_CASE   )






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      None:
          _snake_case							=				self.rear.prev

          # All nodes other than self.head are guaranteed to have non-None previous
          assert previous is not None

          _snake_case							=				node
          _snake_case							=				previous
          _snake_case							=				node
          _snake_case							=				self.rear






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      DoubleLinkedListNode[T, U] | None:

          if node.prev is None or node.next is None:
               return None

          _snake_case							=				node.next
          _snake_case							=				node.prev
          _snake_case							=				None
          _snake_case							=				None
          return node







class        _a    (						Generic[T, U]							):
     SCREAMING_SNAKE_CASE_							:   dict[Callable[[T], U], LRUCache[T, U]]       =						{}






     def __init__(			self ,_SCREAMING_SNAKE_CASE   )							->      str:
          _snake_case							=				DoubleLinkedList()
          _snake_case							=				capacity
          _snake_case							=				0
          _snake_case							=				0
          _snake_case							=				0
          _snake_case							=				{}






     def __repr__(			self   )							->      str:
          return (
              f"""CacheInfo(hits={self.hits}, misses={self.miss}, """
              f"""capacity={self.capacity}, current size={self.num_keys})"""
          )






     def __contains__(			self ,_SCREAMING_SNAKE_CASE   )							->      bool:
          return key in self.cache






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE   )							->      U | None:
          # Note: pythonic interface would throw KeyError rather than return None

          if key in self.cache:
               self.hits += 1
               _snake_case							=				self.cache[key]
               _snake_case							=				self.list.remove(self.cache[key]   )
               assert node == value_node

               # node is guaranteed not None because it is in self.cache
               assert node is not None
               self.list.add(_SCREAMING_SNAKE_CASE   )
               return node.val
          self.miss += 1
          return None






     def     _lowercase     (			self ,_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )							->      None:

          if key not in self.cache:
               if self.num_keys >= self.capacity:
                    # delete first node (oldest) when over capacity
                    _snake_case							=				self.list.head.next

                    # guaranteed to have a non-None first node when num_keys > 0
                    # explain to type checker via assertions
                    assert first_node is not None
                    assert first_node.key is not None
                    assert (
                        self.list.remove(_SCREAMING_SNAKE_CASE   ) is not None
                    )  # node guaranteed to be in list assert node.key is not None

                    del self.cache[first_node.key]
                    self.num_keys -= 1
               _snake_case							=				DoubleLinkedListNode(_SCREAMING_SNAKE_CASE ,_SCREAMING_SNAKE_CASE   )
               self.list.add(self.cache[key]   )
               self.num_keys += 1

          else:
               # bump node to the end of the list, update value
               _snake_case							=				self.list.remove(self.cache[key]   )
               assert node is not None  # node guaranteed to be in list
               _snake_case							=				value
               self.list.add(_SCREAMING_SNAKE_CASE   )






     @classmethod
     def     _lowercase     (			cls ,_SCREAMING_SNAKE_CASE = 128   )							->      Callable[[Callable[[T], U]], Callable[..., U]]:

          def cache_decorator_inner(_SCREAMING_SNAKE_CASE   ) -> Callable[..., U]:
               def cache_decorator_wrapper(*_SCREAMING_SNAKE_CASE   ) -> U:
                    if func not in cls.decorator_function_to_instance_map:
                         _snake_case							=				LRUCache(_SCREAMING_SNAKE_CASE   )

                    _snake_case							=				cls.decorator_function_to_instance_map[func].get(args[0]   )
                    if result is None:
                         _snake_case							=				func(*_SCREAMING_SNAKE_CASE   )
                         cls.decorator_function_to_instance_map[func].put(args[0] ,_SCREAMING_SNAKE_CASE   )
                    return result

               def cache_info() -> LRUCache[T, U]:
                    return cls.decorator_function_to_instance_map[func]

               setattr(_SCREAMING_SNAKE_CASE ,"cache_info" ,_SCREAMING_SNAKE_CASE   )  # noqa: B010

               return cache_decorator_wrapper

          return cache_decorator_inner


if __name__ == "__main__":
   import doctest

   doctest.testmod()


