import json
import os
import shutil
import sys
import tempfile
import unittest
import unittest.mock as mock
from pathlib import Path

from huggingface_hub import HfFolder, delete_repo
from requests.exceptions import HTTPError

from transformers import AutoConfig, BertConfig, GPTaConfig
from transformers.configuration_utils import PretrainedConfig
from transformers.testing_utils import TOKEN, USER, is_staging_test


sys.path.append(str(Path(__file__).parent.parent / 'utils'))

from test_module.custom_configuration import CustomConfig  # noqa E402


SCREAMING_SNAKE_CASE_:					Union[str, Any]  ={
    '''return_dict''': False,
    '''output_hidden_states''': True,
    '''output_attentions''': True,
    '''torchscript''': True,
    '''torch_dtype''': '''float16''',
    '''use_bfloat16''': True,
    '''tf_legacy_loss''': True,
    '''pruned_heads''': {'''a''': 1},
    '''tie_word_embeddings''': False,
    '''is_decoder''': True,
    '''cross_attention_hidden_size''': 1_28,
    '''add_cross_attention''': True,
    '''tie_encoder_decoder''': True,
    '''max_length''': 50,
    '''min_length''': 3,
    '''do_sample''': True,
    '''early_stopping''': True,
    '''num_beams''': 3,
    '''num_beam_groups''': 3,
    '''diversity_penalty''': 0.5,
    '''temperature''': 2.0,
    '''top_k''': 10,
    '''top_p''': 0.7,
    '''typical_p''': 0.2,
    '''repetition_penalty''': 0.8,
    '''length_penalty''': 0.8,
    '''no_repeat_ngram_size''': 5,
    '''encoder_no_repeat_ngram_size''': 5,
    '''bad_words_ids''': [1, 2, 3],
    '''num_return_sequences''': 3,
    '''chunk_size_feed_forward''': 5,
    '''output_scores''': True,
    '''return_dict_in_generate''': True,
    '''forced_bos_token_id''': 2,
    '''forced_eos_token_id''': 3,
    '''remove_invalid_values''': True,
    '''architectures''': ['''BertModel'''],
    '''finetuning_task''': '''translation''',
    '''id2label''': {0: '''label'''},
    '''label2id''': {'''label''': '''0'''},
    '''tokenizer_class''': '''BertTokenizerFast''',
    '''prefix''': '''prefix''',
    '''bos_token_id''': 6,
    '''pad_token_id''': 7,
    '''eos_token_id''': 8,
    '''sep_token_id''': 9,
    '''decoder_start_token_id''': 10,
    '''exponential_decay_length_penalty''': (5, 1.01),
    '''suppress_tokens''': [0, 1],
    '''begin_suppress_tokens''': 2,
    '''task_specific_params''': {'''translation''': '''some_params'''},
    '''problem_type''': '''regression''',
}




@is_staging_test
class 	__A				(			unittest.TestCase     ):



    @classmethod
    def  _lowercase						(cls   :   str			):
         UpperCAmelCase_	   =       TOKEN
         HfFolder.save_token(__a			)



    @classmethod
    def  _lowercase						(cls   :   List[str]			):




         try:
              delete_repo(token=cls._token       ,      repo_id="test-config"			)
         except HTTPError:
              pass

         try:
              delete_repo(token=cls._token       ,      repo_id="valid_org/test-config-org"			)
         except HTTPError:
              pass

         try:
              delete_repo(token=cls._token       ,      repo_id="test-dynamic-config"			)
         except HTTPError:
              pass



    def  _lowercase						(self   :   List[Any]			):
         UpperCAmelCase_	   =       BertConfig(
             vocab_size=99       ,      hidden_size=32       ,      num_hidden_layers=5       ,      num_attention_heads=4       ,      intermediate_size=37			)
         config.push_to_hub("test-config"       ,      use_auth_token=self._token			)

         UpperCAmelCase_	   =       BertConfig.from_pretrained(f"""{USER}/test-config"""			)
         for k, v in config.to_dict().items():
              if k != "transformers_version":
                   self.assertEqual(__a       ,      getattr(__a       ,      __a			)			)

        # Reset repo
         delete_repo(token=self._token       ,      repo_id="test-config"			)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
              config.save_pretrained(__a       ,      repo_id="test-config"       ,      push_to_hub=__a       ,      use_auth_token=self._token			)

         UpperCAmelCase_	   =       BertConfig.from_pretrained(f"""{USER}/test-config"""			)
         for k, v in config.to_dict().items():
              if k != "transformers_version":
                   self.assertEqual(__a       ,      getattr(__a       ,      __a			)			)



    def  _lowercase						(self   :   str			):
         UpperCAmelCase_	   =       BertConfig(
             vocab_size=99       ,      hidden_size=32       ,      num_hidden_layers=5       ,      num_attention_heads=4       ,      intermediate_size=37			)
         config.push_to_hub("valid_org/test-config-org"       ,      use_auth_token=self._token			)

         UpperCAmelCase_	   =       BertConfig.from_pretrained("valid_org/test-config-org"			)
         for k, v in config.to_dict().items():
              if k != "transformers_version":
                   self.assertEqual(__a       ,      getattr(__a       ,      __a			)			)

        # Reset repo
         delete_repo(token=self._token       ,      repo_id="valid_org/test-config-org"			)

         # Push to hub via save_pretrained
         with tempfile.TemporaryDirectory() as tmp_dir:
              config.save_pretrained(
                  __a       ,      repo_id="valid_org/test-config-org"       ,      push_to_hub=__a       ,      use_auth_token=self._token			)

         UpperCAmelCase_	   =       BertConfig.from_pretrained("valid_org/test-config-org"			)
         for k, v in config.to_dict().items():
              if k != "transformers_version":
                   self.assertEqual(__a       ,      getattr(__a       ,      __a			)			)






    def  _lowercase						(self   :   Optional[int]			):
         CustomConfig.register_for_auto_class()
         UpperCAmelCase_	   =       CustomConfig(attribute=42			)

         config.push_to_hub("test-dynamic-config"       ,      use_auth_token=self._token			)

         # This has added the proper auto_map field to the config
         self.assertDictEqual(config.auto_map       ,      {"AutoConfig": "custom_configuration.CustomConfig"}			)

         UpperCAmelCase_	   =       AutoConfig.from_pretrained(f"""{USER}/test-dynamic-config"""       ,      trust_remote_code=__a			)
         # Can't make an isinstance check because the new_config is from the FakeConfig class of a dynamic module
         self.assertEqual(new_config.__class__.__name__       ,      "CustomConfig"			)
         self.assertEqual(new_config.attribute       ,      42			)




class 	__A				(			unittest.TestCase     ):



    def  _lowercase						(self   :   Optional[Any]			):
         UpperCAmelCase_	   =       GPTaConfig()

         # attempt to modify each of int/float/bool/str config records and verify they were updated
         UpperCAmelCase_	   =       c.n_embd + 1  # int
         UpperCAmelCase_	   =       c.resid_pdrop + 1.0  # float
         UpperCAmelCase_	   =       not c.scale_attn_weights  # bool
         UpperCAmelCase_	   =       c.summary_type + "foo"  # str
         c.update_from_string(
             f"""n_embd={n_embd},resid_pdrop={resid_pdrop},scale_attn_weights={scale_attn_weights},summary_type={summary_type}"""			)
         self.assertEqual(__a       ,      c.n_embd       ,      "mismatch for key: n_embd"			)
         self.assertEqual(__a       ,      c.resid_pdrop       ,      "mismatch for key: resid_pdrop"			)
         self.assertEqual(__a       ,      c.scale_attn_weights       ,      "mismatch for key: scale_attn_weights"			)
         self.assertEqual(__a       ,      c.summary_type       ,      "mismatch for key: summary_type"			)



    def  _lowercase						(self   :   Any			):
         UpperCAmelCase_	   =       PretrainedConfig()
         UpperCAmelCase_	   =       [key for key in base_config.__dict__ if key not in config_common_kwargs]
         # If this part of the test fails, you have arguments to addin config_common_kwargs above.
         self.assertListEqual(
             __a       ,      ["is_encoder_decoder", "_name_or_path", "_commit_hash", "transformers_version"]			)
         UpperCAmelCase_	   =       [key for key, value in config_common_kwargs.items() if value == getattr(__a       ,      __a			)]
         if len(__a			) > 0:
              raise ValueError(
                  "The following keys are set with the default values in"
                  " `test_configuration_common.config_common_kwargs` pick another value for them:"
                  f""" {", ".join(__a			)}."""			)



    def  _lowercase						(self   :   Dict			):




         with self.assertRaises(__a			):
              # config is in subfolder, the following should not work without specifying the subfolder
              UpperCAmelCase_	   =       BertConfig.from_pretrained("hf-internal-testing/tiny-random-bert-subfolder"			)

         UpperCAmelCase_	   =       BertConfig.from_pretrained("hf-internal-testing/tiny-random-bert-subfolder"       ,      subfolder="bert"			)

         self.assertIsNotNone(__a			)



    def  _lowercase						(self   :   Optional[int]			):
         UpperCAmelCase_	   =       mock.Mock()
         UpperCAmelCase_	   =       500
         UpperCAmelCase_	   =       {}
         UpperCAmelCase_	   =       HTTPError
         UpperCAmelCase_	   =       {}

         # Download this model to make sure it's in the cache.
         UpperCAmelCase_	   =       BertConfig.from_pretrained("hf-internal-testing/tiny-random-bert"			)

         # Under the mock environment we get a 500 error when trying to reach the model.
         with mock.patch("requests.Session.request"       ,      return_value=__a			) as mock_head:
              UpperCAmelCase_	   =       BertConfig.from_pretrained("hf-internal-testing/tiny-random-bert"			)
              # This check we did call the fake head request
              mock_head.assert_called()



    def  _lowercase						(self   :   List[str]			):
         UpperCAmelCase_	   =       BertConfig.from_pretrained(
             "https://huggingface.co/hf-internal-testing/tiny-random-bert/resolve/main/config.json"			)



    def  _lowercase						(self   :   str			):
         UpperCAmelCase_	   =       AutoConfig.from_pretrained("bert-base-cased"			)
         UpperCAmelCase_	   =       ["config.4.0.0.json"]

         with tempfile.TemporaryDirectory() as tmp_dir:
              configuration.save_pretrained(__a			)
              UpperCAmelCase_	   =       2
              json.dump(configuration.to_dict()       ,      open(os.path.join(__a       ,      "config.4.0.0.json"			)       ,      "w"			)			)

              # This should pick the new configuration file as the version of Transformers is > 4.0.0
              UpperCAmelCase_	   =       AutoConfig.from_pretrained(__a			)
              self.assertEqual(new_configuration.hidden_size       ,      2			)

              # Will need to be adjusted if we reach v42 and this test is still here.
              # Should pick the old configuration file as the version of Transformers is < 4.42.0
              UpperCAmelCase_	   =       ["config.42.0.0.json"]
              UpperCAmelCase_	   =       768
              configuration.save_pretrained(__a			)
              shutil.move(os.path.join(__a       ,      "config.4.0.0.json"			)       ,      os.path.join(__a       ,      "config.42.0.0.json"			)			)
              UpperCAmelCase_	   =       AutoConfig.from_pretrained(__a			)
              self.assertEqual(new_configuration.hidden_size       ,      768			)






    def  _lowercase						(self   :   Optional[Any]			):
         UpperCAmelCase_	   =       "hf-internal-testing/test-two-configs"

         import transformers as new_transformers

         UpperCAmelCase_	   =       "v4.0.0"
         UpperCAmelCase_	   =       new_transformers.models.auto.AutoConfig.from_pretrained(
             __a       ,      return_unused_kwargs=__a			)
         self.assertEqual(new_configuration.hidden_size       ,      2			)
         # This checks `_configuration_file` ia not kept in the kwargs by mistake.
         self.assertDictEqual(__a       ,      {}			)

         # Testing an older version by monkey-patching the version in the module it's used.
         import transformers as old_transformers

         UpperCAmelCase_	   =       "v3.0.0"
         UpperCAmelCase_	   =       old_transformers.models.auto.AutoConfig.from_pretrained(__a			)
         self.assertEqual(old_configuration.hidden_size       ,      768			)





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


SCREAMING_SNAKE_CASE_:					Dict  =logging.get_logger(__name__)




class 	__A				(			UpperCamelCase__     ):



    def  _lowercase						(self   :   Dict       ,      __a   :   Any			):
         if isinstance(__a       ,      __a			):
              UpperCAmelCase_	   =       [label.strip() for label in labels.split(","			) if label.strip()]
         return labels






    def __call__(self   :   Union[str, Any]       ,      __a   :   Optional[int]       ,      __a   :   Optional[int]       ,      __a   :   int			):
         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			):
              UpperCAmelCase_	   =       [sequences]

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

         return sequence_pairs, sequences




@add_end_docstrings(UpperCamelCase__     )
class 	__A				(			UpperCamelCase__     ):



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



    def  _lowercase						(self   :   Any       ,      __a   :   Any       ,      __a   :   int=True       ,      __a   :   Dict=True       ,      __a   :   Any=TruncationStrategy.ONLY_FIRST       ,      **__a   :   Tuple			):
         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(
                  __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.

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

         return inputs



    def  _lowercase						(self   :   List[str]       ,      **__a   :   Tuple			):
         if kwargs.get("multi_class"       ,      __a			) 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   :   Tuple       ,      __a   :   Union[str, List[str]]       ,      *__a   :   Optional[Any]       ,      **__a   :   Tuple       ,      ):
         if len(__a			) == 0:
              pass
         elif len(__a			) == 1 and "candidate_labels" not in kwargs:
              UpperCAmelCase_	   =       args[0]
         else:
              raise ValueError(f"""Unable to understand extra arguments {args}"""			)

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



    def  _lowercase						(self   :   Optional[int]       ,      __a   :   Optional[Any]       ,      __a   :   List[str]=None       ,      __a   :   Any="This example is {}."			):
         UpperCAmelCase_   ,			UpperCAmelCase_	   =       self._args_parser(__a       ,      __a       ,      __a			)

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

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



    def  _lowercase						(self   :   List[str]       ,      __a   :   Any			):
         UpperCAmelCase_	   =       inputs["candidate_label"]
         UpperCAmelCase_	   =       inputs["sequence"]
         UpperCAmelCase_	   =       {k: inputs[k] for k in self.tokenizer.model_input_names}
         UpperCAmelCase_	   =       self.model(**__a			)

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






    def  _lowercase						(self   :   Optional[Any]       ,      __a   :   List[str]       ,      __a   :   Tuple=False			):
         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(__a			)
         UpperCAmelCase_	   =       N // n
         UpperCAmelCase_	   =       logits.reshape((num_sequences, n, -1)			)

         if multi_label or len(__a			) == 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(__a			) / np.exp(__a			).sum(-1       ,      keepdims=__a			)
              UpperCAmelCase_	   =       scores[..., 1]
         else:
              # softmax the "entailment" logits over all candidate labels
              UpperCAmelCase_	   =       reshaped_outputs[..., self.entailment_id]
              UpperCAmelCase_	   =       np.exp(__a			) / np.exp(__a			).sum(-1       ,      keepdims=__a			)

         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(),
         }




