
"""simple docstring"""







import unittest

from transformers import AutoTokenizer, FalconConfig, is_torch_available
from transformers.testing_utils import require_torch, slow, torch_device

from ...generation.test_utils import GenerationTesterMixin
from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, ids_tensor, random_attention_mask
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
						import torch

						from transformers import (
						    FalconForCausalLM,
						    FalconForQuestionAnswering,
						    FalconForSequenceClassification,
						    FalconForTokenClassification,
						    FalconModel,
						)







class 							__magic_name__     :





					def __init__(				self   :							Optional[int]							,						snake_case_   :							str							,						snake_case_   :							List[Any]=3							,						snake_case_   :							Dict=7							,						snake_case_   :							str=True							,						snake_case_   :							int=True							,						snake_case_   :							List[Any]=False							,						snake_case_   :							Optional[Any]=True							,						snake_case_   :							str=99							,						snake_case_   :							str=32							,						snake_case_   :							str=5							,						snake_case_   :							Tuple=4							,						snake_case_   :							Optional[int]=37							,						snake_case_   :							Dict="gelu"							,						snake_case_   :							List[Any]=0.1							,						snake_case_   :							Optional[Any]=0.1							,						snake_case_   :							Union[str, Any]=512							,						snake_case_   :							List[Any]=16							,						snake_case_   :							str=2							,						snake_case_   :							Optional[Any]=0.02							,						snake_case_   :							Optional[Any]=3							,						snake_case_   :							Any=4							,						snake_case_   :							Optional[Any]=None							,						):
												__snake_case     			=       parent
												__snake_case     			=       batch_size
												__snake_case     			=       seq_length
												__snake_case     			=       is_training
												__snake_case     			=       use_input_mask
												__snake_case     			=       use_token_type_ids
												__snake_case     			=       use_labels
												__snake_case     			=       vocab_size
												__snake_case     			=       hidden_size
												__snake_case     			=       num_hidden_layers
												__snake_case     			=       num_attention_heads
												__snake_case     			=       intermediate_size
												__snake_case     			=       hidden_act
												__snake_case     			=       hidden_dropout_prob
												__snake_case     			=       attention_probs_dropout_prob
												__snake_case     			=       max_position_embeddings
												__snake_case     			=       type_vocab_size
												__snake_case     			=       type_sequence_label_size
												__snake_case     			=       initializer_range
												__snake_case     			=       num_labels
												__snake_case     			=       num_choices
												__snake_case     			=       scope





					def 							lowerCAmelCase      (				self   :							Optional[Any]    ):
												__snake_case     			=       ids_tensor([self.batch_size, self.seq_length]							,						self.vocab_size    )

												__snake_case     			=       None
												if self.use_input_mask:
																			__snake_case     			=       random_attention_mask([self.batch_size, self.seq_length]    )

												__snake_case     			=       None

												__snake_case     			=       None
												__snake_case     			=       None
												__snake_case     			=       None
												if self.use_labels:
																			__snake_case     			=       ids_tensor([self.batch_size]							,						self.type_sequence_label_size    )
																			__snake_case     			=       ids_tensor([self.batch_size, self.seq_length]							,						self.num_labels    )
																			__snake_case     			=       ids_tensor([self.batch_size]							,						self.num_choices    )

												__snake_case     			=       self.get_config()

												return config, input_ids, token_type_ids, input_mask, sequence_labels, token_labels, choice_labels





					def 							lowerCAmelCase      (				self   :							List[str]    ):
												return FalconConfig(
												    vocab_size=self.vocab_size							,						hidden_size=self.hidden_size							,						num_hidden_layers=self.num_hidden_layers							,						num_attention_heads=self.num_attention_heads							,						intermediate_size=self.intermediate_size							,						hidden_act=self.hidden_act							,						hidden_dropout_prob=self.hidden_dropout_prob							,						attention_probs_dropout_prob=self.attention_probs_dropout_prob							,						max_position_embeddings=self.max_position_embeddings							,						type_vocab_size=self.type_vocab_size							,						is_decoder=snake_case_							,						initializer_range=self.initializer_range							,						pad_token_id=1							,						new_decoder_architecture=snake_case_							,						)





					def 							lowerCAmelCase      (				self   :							Dict							,						snake_case_   :							List[Any]							,						snake_case_   :							Any							,						snake_case_   :							Dict							,						snake_case_   :							str							,						snake_case_   :							List[Any]							,						snake_case_   :							Optional[int]							,						snake_case_   :							Union[str, Any]    ):
												__snake_case     			=       FalconModel(config=snake_case_    )
												model.to(snake_case_    )
												model.eval()
												__snake_case     			=       model(snake_case_							,						attention_mask=snake_case_    )
												__snake_case     			=       model(snake_case_    )
												self.parent.assertEqual(result.last_hidden_state.shape							,						(self.batch_size, self.seq_length, self.hidden_size)    )





					def 							lowerCAmelCase      (				self   :							Optional[Any]							,						snake_case_   :							Dict							,						snake_case_   :							str							,						snake_case_   :							str							,						snake_case_   :							Union[str, Any]							,						snake_case_   :							str							,						snake_case_   :							Any							,						snake_case_   :							Dict							,						snake_case_   :							Dict							,						snake_case_   :							Dict							,						):
												__snake_case     			=       True
												__snake_case     			=       FalconModel(snake_case_    )
												model.to(snake_case_    )
												model.eval()
												__snake_case     			=       model(
												    snake_case_							,						attention_mask=snake_case_							,						encoder_hidden_states=snake_case_							,						encoder_attention_mask=snake_case_							,						)
												__snake_case     			=       model(
												    snake_case_							,						attention_mask=snake_case_							,						encoder_hidden_states=snake_case_							,						)
												__snake_case     			=       model(snake_case_							,						attention_mask=snake_case_    )
												self.parent.assertEqual(result.last_hidden_state.shape							,						(self.batch_size, self.seq_length, self.hidden_size)    )





					def 							lowerCAmelCase      (				self   :							Tuple							,						snake_case_   :							Union[str, Any]							,						snake_case_   :							Optional[Any]							,						snake_case_   :							List[Any]							,						snake_case_   :							Optional[Any]							,						snake_case_   :							Any							,						snake_case_   :							Dict							,						snake_case_   :							int							,						snake_case_   :							List[Any]							,						snake_case_   :							str							,						):
												__snake_case     			=       FalconForCausalLM(config=snake_case_    )
												model.to(snake_case_    )
												model.eval()
												__snake_case     			=       model(snake_case_							,						attention_mask=snake_case_							,						labels=snake_case_    )
												self.parent.assertEqual(result.logits.shape							,						(self.batch_size, self.seq_length, self.vocab_size)    )





					def 							lowerCAmelCase      (				self   :							List[Any]							,						snake_case_   :							str							,						snake_case_   :							Union[str, Any]							,						snake_case_   :							Dict							,						snake_case_   :							List[str]							,						snake_case_   :							int							,						snake_case_   :							List[str]							,						snake_case_   :							Union[str, Any]							,						snake_case_   :							int							,						snake_case_   :							Optional[Any]							,						):
												__snake_case     			=       True
												__snake_case     			=       True
												__snake_case     			=       FalconForCausalLM(config=snake_case_    )
												model.to(snake_case_    )
												model.eval()

												# first forward pass
												__snake_case     			=       model(
												    snake_case_							,						attention_mask=snake_case_							,						encoder_hidden_states=snake_case_							,						encoder_attention_mask=snake_case_							,						use_cache=snake_case_							,						)
												__snake_case     			=       outputs.past_key_values

												# create hypothetical multiple next token and extent to next_input_ids
												__snake_case     			=       ids_tensor((self.batch_size, 3)							,						config.vocab_size    )
												__snake_case     			=       ids_tensor((self.batch_size, 3)							,						vocab_size=2    )

												# append to next input_ids and
												__snake_case     			=       torch.cat([input_ids, next_tokens]							,						dim=-1    )
												__snake_case     			=       torch.cat([input_mask, next_mask]							,						dim=-1    )

												__snake_case     			=       model(
												    snake_case_							,						attention_mask=snake_case_							,						encoder_hidden_states=snake_case_							,						encoder_attention_mask=snake_case_							,						output_hidden_states=snake_case_							,						)['hidden_states'][0]
												__snake_case     			=       model(
												    snake_case_							,						attention_mask=snake_case_							,						encoder_hidden_states=snake_case_							,						encoder_attention_mask=snake_case_							,						past_key_values=snake_case_							,						output_hidden_states=snake_case_							,						)['hidden_states'][0]

												# select random slice
												__snake_case     			=       ids_tensor((1,)							,						output_from_past.shape[-1]    ).item()
												__snake_case     			=       output_from_no_past[:, -3:, random_slice_idx].detach()
												__snake_case     			=       output_from_past[:, :, random_slice_idx].detach()

												self.parent.assertTrue(output_from_past_slice.shape[1] == next_tokens.shape[1]    )

												# test that outputs are equal for slice
												self.parent.assertTrue(torch.allclose(snake_case_							,						snake_case_							,						atol=1e-3    )    )





					def 							lowerCAmelCase      (				self   :							int    ):
												__snake_case     			=       self.prepare_config_and_inputs()
												(
												    __snake_case       
												)     			=       config_and_inputs
												__snake_case     			=       {'input_ids': input_ids, 'attention_mask': input_mask}
												return config, inputs_dict







@require_torch
class 							__magic_name__     (		lowercase__      ,		lowercase__      ,		lowercase__      ,		unittest.TestCase   ):
					_SCREAMING_SNAKE_CASE       :					Union[str, Any]     							=   (
					    (
					        FalconModel,
					        FalconForCausalLM,
					        FalconForSequenceClassification,
					        FalconForTokenClassification,
					        FalconForQuestionAnswering,
					    )
					    if is_torch_available()
					    else ()
					)
					_SCREAMING_SNAKE_CASE       :					Tuple     							=   (FalconForCausalLM,) if is_torch_available() else ()
					_SCREAMING_SNAKE_CASE       :					List[Any]     							=   (
					    {
					        'feature-extraction': FalconModel,
					        'text-classification': FalconForSequenceClassification,
					        'text-generation': FalconForCausalLM,
					        'question-answering': FalconForQuestionAnswering,
					        'token-classification': FalconForTokenClassification,
					        'zero-shot': FalconForSequenceClassification,
					    }
					    if is_torch_available()
					    else {}
					)
					_SCREAMING_SNAKE_CASE       :					int     							=   False
					_SCREAMING_SNAKE_CASE       :					Union[str, Any]     							=   False





					def 							lowerCAmelCase      (				self   :							Optional[Any]    ):
												__snake_case     			=       FalconModelTester(self    )
												__snake_case     			=       ConfigTester(self							,						config_class=snake_case_							,						hidden_size=37    )





					def 							lowerCAmelCase      (				self   :							Tuple    ):
												self.config_tester.run_common_tests()





					def 							lowerCAmelCase      (				self   :							Union[str, Any]    ):
												__snake_case     			=       self.model_tester.prepare_config_and_inputs()
												self.model_tester.create_and_check_model(*snake_case_    )





					def 							lowerCAmelCase      (				self   :							Optional[int]    ):
												__snake_case     			=       self.model_tester.prepare_config_and_inputs()
												for alibi in [True, False]:
																			__snake_case     			=       alibi
																			self.model_tester.create_and_check_model(snake_case_							,						*snake_case_    )





					def 							lowerCAmelCase      (				self   :							List[Any]    ):
												__snake_case     			=       self.model_tester.prepare_config_and_inputs_for_common()
												__snake_case     			=       3
												__snake_case     			=       input_dict['input_ids']
												__snake_case     			=       input_ids.ne(1    ).to(snake_case_    )
												__snake_case     			=       ids_tensor([self.model_tester.batch_size]							,						self.model_tester.type_sequence_label_size    )
												__snake_case     			=       FalconForSequenceClassification(snake_case_    )
												model.to(snake_case_    )
												model.eval()
												__snake_case     			=       model(snake_case_							,						attention_mask=snake_case_							,						labels=snake_case_    )
												self.assertEqual(result.logits.shape							,						(self.model_tester.batch_size, self.model_tester.num_labels)    )





					def 							lowerCAmelCase      (				self   :							Dict    ):
												__snake_case     			=       self.model_tester.prepare_config_and_inputs_for_common()
												__snake_case     			=       3
												__snake_case     			=       'single_label_classification'
												__snake_case     			=       input_dict['input_ids']
												__snake_case     			=       input_ids.ne(1    ).to(snake_case_    )
												__snake_case     			=       ids_tensor([self.model_tester.batch_size]							,						self.model_tester.type_sequence_label_size    )
												__snake_case     			=       FalconForSequenceClassification(snake_case_    )
												model.to(snake_case_    )
												model.eval()
												__snake_case     			=       model(snake_case_							,						attention_mask=snake_case_							,						labels=snake_case_    )
												self.assertEqual(result.logits.shape							,						(self.model_tester.batch_size, self.model_tester.num_labels)    )





					def 							lowerCAmelCase      (				self   :							List[Any]    ):
												__snake_case     			=       self.model_tester.prepare_config_and_inputs_for_common()
												__snake_case     			=       input_dict['input_ids']
												__snake_case     			=       FalconForCausalLM(snake_case_    )
												model.to(snake_case_    )
												model.eval()
												__snake_case     			=       model(snake_case_							,						use_cache=snake_case_    )
												__snake_case     			=       input_ids.shape[0]
												__snake_case     			=       model._convert_to_rw_cache(result.past_key_values    )
												__snake_case     			=       model._convert_cache_to_standard_format(snake_case_							,						snake_case_    )
												for layer in range(len(snake_case_    )    ):
																			for tensor_idx in range(2    ):
																										self.assertTrue(rw_cache[layer][tensor_idx].ndim == 3    )
																										self.assertTrue(result.past_key_values[layer][tensor_idx].ndim == 4    )
																										self.assertTrue(
																										    torch.all(result.past_key_values[layer][tensor_idx] == standard_cache[layer][tensor_idx]    )    )





					def 							lowerCAmelCase      (				self   :							Tuple    ):
												__snake_case     			=       self.model_tester.prepare_config_and_inputs_for_common()
												__snake_case     			=       3
												__snake_case     			=       'multi_label_classification'
												__snake_case     			=       input_dict['input_ids']
												__snake_case     			=       input_ids.ne(1    ).to(snake_case_    )
												__snake_case     			=       ids_tensor(
												    [self.model_tester.batch_size, config.num_labels]							,						self.model_tester.type_sequence_label_size    ).to(torch.float    )
												__snake_case     			=       FalconForSequenceClassification(snake_case_    )
												model.to(snake_case_    )
												model.eval()
												__snake_case     			=       model(snake_case_							,						attention_mask=snake_case_							,						labels=snake_case_    )
												self.assertEqual(result.logits.shape							,						(self.model_tester.batch_size, self.model_tester.num_labels)    )





					def 							lowerCAmelCase      (				self   :							List[str]    ):







												for model_class in self.all_generative_model_classes:
																			__snake_case     			=       self.model_tester.prepare_config_and_inputs_for_common()

																			# If it doesn't support cache, pass the test
																			if not hasattr(snake_case_							,						"use_cache"    ):
																										return

																			__snake_case     			=       model_class(snake_case_    ).to(snake_case_    )
																			if "use_cache" not in inputs:
																										__snake_case     			=       True
																			__snake_case     			=       model(**snake_case_    )

																			# If "past_key_values" is not returned, pass the test (e.g. RWKV uses a different cache name and format)
																			if "past_key_values" not in outputs:
																										return

																			__snake_case     			=       (
																			    getattr(snake_case_							,						"decoder_layers"							,						snake_case_    )
																			    or getattr(snake_case_							,						"num_decoder_layers"							,						snake_case_    )
																			    or config.num_hidden_layers
																			)
																			__snake_case     			=       getattr(snake_case_							,						"num_kv_heads"							,						config.num_attention_heads    )
																			__snake_case     			=       getattr(snake_case_							,						"d_model"							,						config.hidden_size    )
																			__snake_case     			=       embed_dim // num_attention_heads

																			__snake_case     			=       outputs['past_key_values']
																			self.assertEqual(len(snake_case_    )							,						snake_case_    )

																			__snake_case     			=       inputs['input_ids'].shape
																			for i in range(snake_case_    ):
																										if config.new_decoder_architecture:
																																	__snake_case     			=       config.num_attention_heads
																										elif config.multi_query:
																																	__snake_case     			=       1
																										self.assertEqual(len(past_kv[0]    )							,						2    )  # K V for the decoder = 2
																										self.assertEqual(
																										    past_kv[i][0].shape							,						(batch_size, num_attention_heads, seq_length, per_head_embed_dim)    )
																										self.assertEqual(
																										    past_kv[i][1].shape							,						(batch_size, num_attention_heads, seq_length, per_head_embed_dim)    )







@require_torch
class 							__magic_name__     (		unittest.TestCase   ):





					@slow
					def 							lowerCAmelCase      (				self   :							Optional[Any]    ):
												__snake_case     			=       AutoTokenizer.from_pretrained("Rocketknight1/falcon-rw-1b"    )
												__snake_case     			=       FalconForCausalLM.from_pretrained("Rocketknight1/falcon-rw-1b"    )
												model.eval()
												model.to(snake_case_    )
												__snake_case     			=       tokenizer("My favorite food is"							,						return_tensors="pt"    ).to(snake_case_    )

												__snake_case     			=       (
												    'My favorite food is pizza. I love it so much that I have a pizza party every year for my birthday.'
												)

												__snake_case     			=       model.generate(**snake_case_							,						do_sample=snake_case_							,						max_new_tokens=19    )
												__snake_case     			=       tokenizer.batch_decode(snake_case_    )[0]

												self.assertEqual(snake_case_							,						snake_case_    )





					@slow
					def 							lowerCAmelCase      (				self   :							List[str]    ):







												for repo in ["Rocketknight1/tiny-random-falcon-7b", "Rocketknight1/tiny-random-falcon-40b"]:
																			__snake_case     			=       AutoTokenizer.from_pretrained(snake_case_    )
																			__snake_case     			=       FalconForCausalLM.from_pretrained(snake_case_    )
																			model.eval()
																			model.to(snake_case_    )
																			__snake_case     			=       tokenizer("My favorite food is"							,						return_tensors="pt"    ).to(snake_case_    )

																			# We just test that these run without errors - the models are randomly initialized
																			# and so the actual text outputs will be garbage
																			model.generate(**snake_case_							,						do_sample=snake_case_							,						max_new_tokens=4    )
																			model.generate(**snake_case_							,						do_sample=snake_case_							,						max_new_tokens=4    )
																			model.generate(**snake_case_							,						num_beams=2							,						max_new_tokens=4    )





					@slow
					def 							lowerCAmelCase      (				self   :							int    ):







												with torch.no_grad():
																			for repo in [
																			    "Rocketknight1/falcon-rw-1b",
																			    "Rocketknight1/tiny-random-falcon-7b",
																			    "Rocketknight1/tiny-random-falcon-40b",
																			]:
																										__snake_case     			=       AutoTokenizer.from_pretrained(snake_case_    )
																										__snake_case     			=       FalconForCausalLM.from_pretrained(snake_case_    )
																										model.eval()
																										model.to(device=snake_case_    )
																										__snake_case     			=       tokenizer("My favorite food is"							,						return_tensors="pt"    ).to(snake_case_    )

																										# Test results are the same with and without cache
																										__snake_case     			=       model.generate(**snake_case_							,						do_sample=snake_case_							,						max_new_tokens=20							,						use_cache=snake_case_    )
																										__snake_case     			=       model.generate(**snake_case_							,						do_sample=snake_case_							,						max_new_tokens=20							,						use_cache=snake_case_    )
																										self.assertTrue((outputs_cache - outputs_no_cache).sum().item() == 0    )


"""simple docstring"""







from collections import UserDict
from typing import List, Union

from ..utils import (
    add_end_docstrings,
    is_tf_available,
    is_torch_available,
    is_vision_available,
    logging,
    requires_backends,
)
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_vision_available():
		from PIL import Image

		from ..image_utils import load_image

if is_torch_available():
		from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING

if is_tf_available():
		from ..models.auto.modeling_tf_auto import TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
		from ..tf_utils import stable_softmax

_SCREAMING_SNAKE_CASE	       =      logging.get_logger(__name__)







@add_end_docstrings(lowercase__   )
class 							__magic_name__     (		lowercase__   ):





					def __init__(				self   :							Tuple							,						**snake_case_   :							str    ):
												super().__init__(**snake_case_    )

												requires_backends(self							,						"vision"    )
												self.check_model_type(
												    TF_MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING
												    if self.framework == "tf"
												    else MODEL_FOR_ZERO_SHOT_IMAGE_CLASSIFICATION_MAPPING    )





					def __call__(				self   :							Optional[Any]							,						snake_case_   :							Union[str, List[str], "Image", List["Image"]]							,						**snake_case_   :							Optional[int]    ):
												return super().__call__(snake_case_							,						**snake_case_    )





					def 							lowerCAmelCase      (				self   :							Tuple							,						**snake_case_   :							Any    ):
												__snake_case     			=       {}
												if "candidate_labels" in kwargs:
																			__snake_case     			=       kwargs["candidate_labels"]
												if "hypothesis_template" in kwargs:
																			__snake_case     			=       kwargs["hypothesis_template"]

												return preprocess_params, {}, {}





					def 							lowerCAmelCase      (				self   :							int							,						snake_case_   :							List[str]							,						snake_case_   :							List[str]=None							,						snake_case_   :							int="This is a photo of {}."    ):
												__snake_case     			=       load_image(snake_case_    )
												__snake_case     			=       self.image_processor(images=[image]							,						return_tensors=self.framework    )
												__snake_case     			=       candidate_labels
												__snake_case     			=       [hypothesis_template.format(snake_case_    ) for x in candidate_labels]
												__snake_case     			=       self.tokenizer(snake_case_							,						return_tensors=self.framework							,						padding=snake_case_    )
												__snake_case     			=       [text_inputs]
												return inputs





					def 							lowerCAmelCase      (				self   :							Any							,						snake_case_   :							Dict    ):
												__snake_case     			=       model_inputs.pop("candidate_labels"    )
												__snake_case     			=       model_inputs.pop("text_inputs"    )
												if isinstance(text_inputs[0]							,						snake_case_    ):
																			__snake_case     			=       text_inputs[0]
												else:
																			# Batching case.
																			__snake_case     			=       text_inputs[0][0]

												__snake_case     			=       self.model(**snake_case_							,						**snake_case_    )

												__snake_case     			=       {
												    "candidate_labels": candidate_labels,
												    "logits": outputs.logits_per_image,
												}
												return model_outputs





					def 							lowerCAmelCase      (				self   :							Optional[int]							,						snake_case_   :							Any    ):
												__snake_case     			=       model_outputs.pop("candidate_labels"    )
												__snake_case     			=       model_outputs["logits"][0]
												if self.framework == "pt":
																			__snake_case     			=       logits.softmax(dim=-1    ).squeeze(-1    )
																			__snake_case     			=       probs.tolist()
																			if not isinstance(snake_case_							,						snake_case_    ):
																										__snake_case     			=       [scores]
												elif self.framework == "tf":
																			__snake_case     			=       stable_softmax(snake_case_							,						axis=-1    )
																			__snake_case     			=       probs.numpy().tolist()
												else:
																			raise ValueError(F'''Unsupported framework: {self.framework}'''    )

												__snake_case     			=       [
												    {"score": score, "label": candidate_label}
												    for score, candidate_label in sorted(zip(snake_case_							,						snake_case_    )							,						key=lambda snake_case_    : -x[0]    )
												]
												return result
