


"""simple docstring"""







import uuid
from typing import Any, Dict, List, Optional, Union

from ..utils import add_end_docstrings, is_tf_available, is_torch_available, logging
from .base import PIPELINE_INIT_ARGS, Pipeline


if is_tf_available():
       import tensorflow as tf

if is_torch_available():
       import torch


lowerCamelCase_       :       str								=  logging.get_logger(__name__)






class 				__A    :






 """simple docstring"""




 def __init__(		self			,   __A = None			,   __A = None			,   __A=None			,   __A=None						) ->					Optional[Any]:
     if not conversation_id:
         a					=uuid.uuida()
     if past_user_inputs is None:
         a					=[]
     if generated_responses is None:
         a					=[]

     a					=conversation_id
     a					=past_user_inputs
     a					=generated_responses
     a					=text


 def __eq__(		self			,   __A						) ->					Tuple:
     if not isinstance(SCREAMING_SNAKE_CASE_			,   SCREAMING_SNAKE_CASE_						):
         return False
     if self.uuid == other.uuid:
         return True
     return (
         self.new_user_input == other.new_user_input
         and self.past_user_inputs == other.past_user_inputs
         and self.generated_responses == other.generated_responses
     )


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A = False						) ->					Any:
     if self.new_user_input:
         if overwrite:
             logger.warning(
                 f'''User input added while unprocessed input was existing: "{self.new_user_input}" was overwritten '''
                 f'''with: "{text}".'''						)
             a					=text
         else:
             logger.warning(
                 f'''User input added while unprocessed input was existing: "{self.new_user_input}" new input '''
                 f'''ignored: "{text}". Set `overwrite` to True to overwrite unprocessed user input'''						)
     else:
         a					=text


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     if self.new_user_input:
         self.past_user_inputs.append(self.new_user_input						)
     a					=None


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A						) ->					int:
     self.generated_responses.append(SCREAMING_SNAKE_CASE_						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Any:
     for user_input, generated_response in zip(self.past_user_inputs			,   self.generated_responses						):
         yield True, user_input
         yield False, generated_response
     if self.new_user_input:
         yield True, self.new_user_input





 def __repr__(		self						) ->					Dict:
     a					=f'''Conversation id: {self.uuid} \n'''
     for is_user, text in self.iter_texts():
         a					="""user""" if is_user else """bot"""
         output += f'''{name} >> {text} \n'''
     return output






@add_end_docstrings(
    _UpperCAmelCase,   r"\n        min_length_for_response (`int`, *optional*, defaults to 32):\n            The minimum length (in number of tokens) for a response.\n        minimum_tokens (`int`, *optional*, defaults to 10):\n            The minimum length of tokens to leave for a response.\n    ",   )
class 				__A    (	_UpperCAmelCase      ):






 """simple docstring"""




 def __init__(		self			,   *__A			,   **__A						) ->					List[str]:
     super().__init__(*SCREAMING_SNAKE_CASE_			,   **SCREAMING_SNAKE_CASE_						)
     if self.tokenizer.pad_token_id is None:
         a					=self.tokenizer.eos_token


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A=None			,   __A=None			,   __A=None			,   **__A						) ->					Dict:
     a					={}
     a					={}
     a					={}

     if min_length_for_response is not None:
         a					=min_length_for_response
     if minimum_tokens is not None:
         a					=minimum_tokens

     if "max_length" in generate_kwargs:
         a					=generate_kwargs["""max_length"""]
         # self.max_length = generate_kwargs.get("max_length", self.model.config.max_length)
     if clean_up_tokenization_spaces is not None:
         a					=clean_up_tokenization_spaces

     if generate_kwargs:
         forward_params.update(SCREAMING_SNAKE_CASE_						)
     return preprocess_params, forward_params, postprocess_params


 def __call__(		self			,   __A			,   __A=0			,   **__A						) ->					Optional[int]:
     a					=super().__call__(SCREAMING_SNAKE_CASE_			,   num_workers=SCREAMING_SNAKE_CASE_			,   **SCREAMING_SNAKE_CASE_						)
     if isinstance(SCREAMING_SNAKE_CASE_			,   SCREAMING_SNAKE_CASE_						) and len(SCREAMING_SNAKE_CASE_						) == 1:
         return outputs[0]
     return outputs


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A=32						) ->					Optional[int]:
     if not isinstance(SCREAMING_SNAKE_CASE_			,   SCREAMING_SNAKE_CASE_						):
         raise ValueError('''ConversationalPipeline, expects Conversation as inputs'''						)
     if conversation.new_user_input is None:
         raise ValueError(
             f'''Conversation with UUID {type(conversation.uuid						)} does not contain new user input to process. '''
             '''Add user inputs with the conversation\'s `add_user_input` method'''						)
     if hasattr(self.tokenizer			,   '''_build_conversation_input_ids'''						):
         a					=self.tokenizer._build_conversation_input_ids(SCREAMING_SNAKE_CASE_						)
     else:
         # If the tokenizer cannot handle conversations, we default to only the old version
         a					=self._legacy_parse_and_tokenize(SCREAMING_SNAKE_CASE_						)

     if self.framework == "pt":
         a					=torch.LongTensor([input_ids]						)
     elif self.framework == "tf":
         a					=tf.constant([input_ids]						)
     return {"input_ids": input_ids, "conversation": conversation}


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A=10			,   **__A						) ->					Tuple:
     a					=generate_kwargs.get('''max_length'''			,   self.model.config.max_length						)

     a					=model_inputs["""input_ids"""].shape[1]
     if max_length - minimum_tokens < n:
         logger.warning(f'''Conversation input is to long ({n}), trimming it to ({max_length} - {minimum_tokens})'''						)
         a					=max_length - minimum_tokens
         a					=model_inputs["""input_ids"""][:, -trim:]
         if "attention_mask" in model_inputs:
             a					=model_inputs["""attention_mask"""][:, -trim:]
     a					=model_inputs.pop('''conversation'''						)
     a					=max_length
     a					=self.model.generate(**SCREAMING_SNAKE_CASE_			,   **SCREAMING_SNAKE_CASE_						)
     if self.model.config.is_encoder_decoder:
         a					=1
     else:
         a					=n
     return {"output_ids": output_ids[:, start_position:], "conversation": conversation}


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A=True						) ->					List[Any]:
     a					=model_outputs["""output_ids"""]
     a					=self.tokenizer.decode(
         output_ids[0]			,   skip_special_tokens=SCREAMING_SNAKE_CASE_			,   clean_up_tokenization_spaces=SCREAMING_SNAKE_CASE_			,   )
     a					=model_outputs["""conversation"""]
     conversation.mark_processed()
     conversation.append_response(SCREAMING_SNAKE_CASE_						)
     return conversation





 def 				SCREAMING_SNAKE_CASE							(		self			,   __A						) ->					List[str]:
     a					=self.tokenizer.eos_token_id
     a					=[]
     for is_user, text in conversation.iter_texts():
         if eos_token_id is not None:
             input_ids.extend(self.tokenizer.encode(SCREAMING_SNAKE_CASE_			,   add_special_tokens=SCREAMING_SNAKE_CASE_						) + [eos_token_id]						)
         else:
             input_ids.extend(self.tokenizer.encode(SCREAMING_SNAKE_CASE_			,   add_special_tokens=SCREAMING_SNAKE_CASE_						)						)

     if len(SCREAMING_SNAKE_CASE_						) > self.tokenizer.model_max_length:
         a					=input_ids[-self.tokenizer.model_max_length :]
     return input_ids



"""simple docstring"""







import inspect
import unittest

from transformers import YolosConfig
from transformers.testing_utils import require_torch, require_vision, slow, torch_device
from transformers.utils import cached_property, is_torch_available, is_vision_available

from ...test_configuration_common import ConfigTester
from ...test_modeling_common import ModelTesterMixin, floats_tensor
from ...test_pipeline_mixin import PipelineTesterMixin


if is_torch_available():
       import torch
       from torch import nn

       from transformers import YolosForObjectDetection, YolosModel
       from transformers.models.yolos.modeling_yolos import YOLOS_PRETRAINED_MODEL_ARCHIVE_LIST


if is_vision_available():
       from PIL import Image

       from transformers import AutoImageProcessor






class 				__A    :






 """simple docstring"""




 def __init__(		self			,   __A			,   __A=13			,   __A=[30, 30]			,   __A=2			,   __A=3			,   __A=True			,   __A=True			,   __A=32			,   __A=5			,   __A=4			,   __A=37			,   __A="gelu"			,   __A=0.1			,   __A=0.1			,   __A=10			,   __A=0.02			,   __A=3			,   __A=None			,   __A=8			,   __A=10			,   ) ->					List[Any]:
     a					=parent
     a					=batch_size
     a					=image_size
     a					=patch_size
     a					=num_channels
     a					=is_training
     a					=use_labels
     a					=hidden_size
     a					=num_hidden_layers
     a					=num_attention_heads
     a					=intermediate_size
     a					=hidden_act
     a					=hidden_dropout_prob
     a					=attention_probs_dropout_prob
     a					=type_sequence_label_size
     a					=initializer_range
     a					=num_labels
     a					=scope
     a					=n_targets
     a					=num_detection_tokens
     # we set the expected sequence length (which is used in several tests)
     # expected sequence length = num_patches + 1 (we add 1 for the [CLS] token) + num_detection_tokens
     a					=(image_size[1] // patch_size) * (image_size[0] // patch_size)
     a					=num_patches + 1 + self.num_detection_tokens


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Optional[Any]:
     a					=floats_tensor([self.batch_size, self.num_channels, self.image_size[0], self.image_size[1]]						)

     a					=None
     if self.use_labels:
         # labels is a list of Dict (each Dict being the labels for a given example in the batch)
         a					=[]
         for i in range(self.batch_size						):
             a					={}
             a					=torch.randint(
                 high=self.num_labels			,   size=(self.n_targets,)			,   device=__A						)
             a					=torch.rand(self.n_targets			,   4			,   device=__A						)
             labels.append(__A						)

     a					=self.get_config()

     return config, pixel_values, labels


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     return YolosConfig(
         image_size=self.image_size			,   patch_size=self.patch_size			,   num_channels=self.num_channels			,   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			,   is_decoder=__A			,   initializer_range=self.initializer_range			,   num_detection_tokens=self.num_detection_tokens			,   num_labels=self.num_labels			,   )


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A			,   __A						) ->					List[Any]:
     a					=YolosModel(config=__A						)
     model.to(__A						)
     model.eval()
     a					=model(__A						)
     self.parent.assertEqual(
         result.last_hidden_state.shape			,   (self.batch_size, self.expected_seq_len, self.hidden_size)						)


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A			,   __A						) ->					Dict:
     a					=YolosForObjectDetection(__A						)
     model.to(__A						)
     model.eval()

     a					=model(pixel_values=__A						)
     a					=model(__A						)

     self.parent.assertEqual(result.logits.shape			,   (self.batch_size, self.num_detection_tokens, self.num_labels + 1)						)
     self.parent.assertEqual(result.pred_boxes.shape			,   (self.batch_size, self.num_detection_tokens, 4)						)

     a					=model(pixel_values=__A			,   labels=__A						)

     self.parent.assertEqual(result.loss.shape			,   ()						)
     self.parent.assertEqual(result.logits.shape			,   (self.batch_size, self.num_detection_tokens, self.num_labels + 1)						)
     self.parent.assertEqual(result.pred_boxes.shape			,   (self.batch_size, self.num_detection_tokens, 4)						)





 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Union[str, Any]:
     a					=self.prepare_config_and_inputs()
     a		,				a		,				a					=config_and_inputs
     a					={'''pixel_values''': pixel_values}
     return config, inputs_dict






@require_torch
class 				__A    (	_SCREAMING_SNAKE_CASE,   _SCREAMING_SNAKE_CASE,   unittest.TestCase      ):






 """simple docstring"""




 __lowerCAmelCase			 =  (YolosModel, YolosForObjectDetection) if is_torch_available() else ()
 __lowerCAmelCase			 =  (
     {"feature-extraction": YolosModel, "object-detection": YolosForObjectDetection} if is_torch_available() else {}
 )

 __lowerCAmelCase			 =  False
 __lowerCAmelCase			 =  False
 __lowerCAmelCase			 =  False
 __lowerCAmelCase			 =  False


 def 				SCREAMING_SNAKE_CASE							(		self			,   __A			,   __A			,   __A=False						) ->					Any:
     a					=super()._prepare_for_class(__A			,   __A			,   return_labels=__A						)

     if return_labels:
         if model_class.__name__ == "YolosForObjectDetection":
             a					=[]
             for i in range(self.model_tester.batch_size						):
                 a					={}
                 a					=torch.ones(
                     size=(self.model_tester.n_targets,)			,   device=__A			,   dtype=torch.long						)
                 a					=torch.ones(
                     self.model_tester.n_targets			,   4			,   device=__A			,   dtype=torch.float						)
                 labels.append(__A						)
             a					=labels

     return inputs_dict


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[str]:
     a					=YolosModelTester(self						)
     a					=ConfigTester(self			,   config_class=__A			,   has_text_modality=__A			,   hidden_size=37						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Dict:
     self.config_tester.run_common_tests()


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[Any]:
     # YOLOS does not use inputs_embeds
     pass


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Optional[Any]:
     a		,				a					=self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
         a					=model_class(__A						)
         self.assertIsInstance(model.get_input_embeddings()			,   (nn.Module)						)
         a					=model.get_output_embeddings()
         self.assertTrue(x is None or isinstance(__A			,   nn.Linear						)						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Optional[int]:
     a		,				a					=self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
         a					=model_class(__A						)
         a					=inspect.signature(model.forward						)
         # signature.parameters is an OrderedDict => so arg_names order is deterministic
         a					=[*signature.parameters.keys()]

         a					=['''pixel_values''']
         self.assertListEqual(arg_names[:1]			,   __A						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Optional[Any]:
     a					=self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_model(*__A						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[Any]:
     a		,				a					=self.model_tester.prepare_config_and_inputs_for_common()
     a					=True

     # in YOLOS, the seq_len is different
     a					=self.model_tester.expected_seq_len
     for model_class in self.all_model_classes:
         a					=True
         a					=False
         a					=True
         a					=model_class(__A						)
         model.to(__A						)
         model.eval()
         with torch.no_grad():
             a					=model(**self._prepare_for_class(__A			,   __A						)						)
         a					=outputs.attentions
         self.assertEqual(len(__A						)			,   self.model_tester.num_hidden_layers						)

         # check that output_attentions also work using config
         del inputs_dict["output_attentions"]
         a					=True
         a					=model_class(__A						)
         model.to(__A						)
         model.eval()
         with torch.no_grad():
             a					=model(**self._prepare_for_class(__A			,   __A						)						)
         a					=outputs.attentions
         self.assertEqual(len(__A						)			,   self.model_tester.num_hidden_layers						)

         self.assertListEqual(
             list(attentions[0].shape[-3:]						)			,   [self.model_tester.num_attention_heads, seq_len, seq_len]			,   )
         a					=len(__A						)

         # Check attention is always last and order is fine
         a					=True
         a					=True
         a					=model_class(__A						)
         model.to(__A						)
         model.eval()
         with torch.no_grad():
             a					=model(**self._prepare_for_class(__A			,   __A						)						)

         a					=1
         self.assertEqual(out_len + added_hidden_states			,   len(__A						)						)

         a					=outputs.attentions

         self.assertEqual(len(__A						)			,   self.model_tester.num_hidden_layers						)
         self.assertListEqual(
             list(self_attentions[0].shape[-3:]						)			,   [self.model_tester.num_attention_heads, seq_len, seq_len]			,   )


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					str:
     def check_hidden_states_output(__A			,   __A			,   __A						):
         a					=model_class(__A						)
         model.to(__A						)
         model.eval()

         with torch.no_grad():
             a					=model(**self._prepare_for_class(__A			,   __A						)						)

         a					=outputs.hidden_states

         a					=getattr(
             self.model_tester			,   '''expected_num_hidden_layers'''			,   self.model_tester.num_hidden_layers + 1						)
         self.assertEqual(len(__A						)			,   __A						)

         # YOLOS has a different seq_length
         a					=self.model_tester.expected_seq_len

         self.assertListEqual(
             list(hidden_states[0].shape[-2:]						)			,   [seq_length, self.model_tester.hidden_size]			,   )

     a		,				a					=self.model_tester.prepare_config_and_inputs_for_common()

     for model_class in self.all_model_classes:
         a					=True
         check_hidden_states_output(__A			,   __A			,   __A						)

         # check that output_hidden_states also work using config
         del inputs_dict["output_hidden_states"]
         a					=True

         check_hidden_states_output(__A			,   __A			,   __A						)


 def 				SCREAMING_SNAKE_CASE							(		self						) ->					int:
     a					=self.model_tester.prepare_config_and_inputs()
     self.model_tester.create_and_check_for_object_detection(*__A						)





 @slow
 def 				SCREAMING_SNAKE_CASE							(		self						) ->					List[Any]:
     for model_name in YOLOS_PRETRAINED_MODEL_ARCHIVE_LIST[:1]:
         a					=YolosModel.from_pretrained(__A						)
         self.assertIsNotNone(__A						)

def _A						(						):







    """simple docstring"""
    a					=Image.open('''./tests/fixtures/tests_samples/COCO/000000039769.png'''		)
    return image






@require_torch
@require_vision
class 				__A    (	unittest.TestCase      ):






 """simple docstring"""




 @cached_property
 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Dict:
     return AutoImageProcessor.from_pretrained('''hustvl/yolos-small'''						) if is_vision_available() else None





 @slow
 def 				SCREAMING_SNAKE_CASE							(		self						) ->					Union[str, Any]:
     a					=YolosForObjectDetection.from_pretrained('''hustvl/yolos-small'''						).to(__A						)

     a					=self.default_image_processor
     a					=prepare_img()
     a					=image_processor(images=__A			,   return_tensors='''pt'''						).to(__A						)

     # forward pass
     with torch.no_grad():
         a					=model(inputs.pixel_values						)

     # verify outputs
     a					=torch.Size((1, 100, 92)						)
     self.assertEqual(outputs.logits.shape			,   __A						)

     a					=torch.tensor(
         [[-24.0_248, -10.3_024, -14.8_290], [-42.0_392, -16.8_200, -27.4_334], [-27.2_743, -11.8_154, -18.7_148]]			,   device=__A			,   )
     a					=torch.tensor(
         [[0.2_559, 0.5_455, 0.4_706], [0.2_989, 0.7_279, 0.1_875], [0.7_732, 0.4_017, 0.4_462]]			,   device=__A						)
     self.assertTrue(torch.allclose(outputs.logits[0, :3, :3]			,   __A			,   atol=1E-4						)						)
     self.assertTrue(torch.allclose(outputs.pred_boxes[0, :3, :3]			,   __A			,   atol=1E-4						)						)

     # verify postprocessing
     a					=image_processor.post_process_object_detection(
         __A			,   threshold=0.3			,   target_sizes=[image.size[::-1]]						)[0]
     a					=torch.tensor([0.9_994, 0.9_790, 0.9_964, 0.9_972, 0.9_861]						).to(__A						)
     a					=[75, 75, 17, 63, 17]
     a					=torch.tensor([335.0_609, 79.3_848, 375.4_216, 187.2_495]						).to(__A						)

     self.assertEqual(len(results['''scores''']						)			,   5						)
     self.assertTrue(torch.allclose(results['''scores''']			,   __A			,   atol=1E-4						)						)
     self.assertSequenceEqual(results['''labels'''].tolist()			,   __A						)
     self.assertTrue(torch.allclose(results['''boxes'''][0, :]			,   __A						)						)