



import unittest

from transformers import SpeechaTextaConfig
from transformers.testing_utils import is_torch_available, require_torch, torch_device

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


if is_torch_available():
      import torch

      from transformers.models.speech_to_text_a.modeling_speech_to_text_a import (
          SpeechaTextaDecoder,
          SpeechaTextaForCausalLM,
      )







@require_torch
class       _snake_case				:






   '''simple docstring'''






   def __init__(    self:     Optional[Any]  ,lowerCamelCase_:     Union[str, Any]  ,lowerCamelCase_:     Union[str, Any]=99  ,lowerCamelCase_:     int=13  ,lowerCamelCase_:     List[Any]=16  ,lowerCamelCase_:     int=7  ,lowerCamelCase_:     Optional[Any]=True  ,lowerCamelCase_:     Optional[Any]=True  ,lowerCamelCase_:     Any=True  ,lowerCamelCase_:     str=False  ,lowerCamelCase_:     str=True  ,lowerCamelCase_:     Union[str, Any]=2  ,lowerCamelCase_:     Optional[Any]=32  ,lowerCamelCase_:     Dict=4  ,lowerCamelCase_:     Optional[int]=4  ,lowerCamelCase_:     Optional[int]=30  ,lowerCamelCase_:     Tuple=0  ,lowerCamelCase_:     List[Any]=1  ,lowerCamelCase_:     List[Any]=2  ,lowerCamelCase_:     Optional[int]=None  ,)	->      List[str]:
       UpperCAmelCase_     :       str  		=					parent
       UpperCAmelCase_     :       int  		=					batch_size
       UpperCAmelCase_     :       Any  		=					decoder_seq_length
       # For common tests
       UpperCAmelCase_     :       Tuple  		=					self.decoder_seq_length
       UpperCAmelCase_     :       List[Any]  		=					is_training
       UpperCAmelCase_     :       Any  		=					use_attention_mask
       UpperCAmelCase_     :       List[Any]  		=					use_labels

       UpperCAmelCase_     :       Tuple  		=					vocab_size
       UpperCAmelCase_     :       Dict  		=					d_model
       UpperCAmelCase_     :       List[Any]  		=					d_model
       UpperCAmelCase_     :       Optional[Any]  		=					decoder_layers
       UpperCAmelCase_     :       List[str]  		=					decoder_layers
       UpperCAmelCase_     :       Optional[int]  		=					decoder_ffn_dim
       UpperCAmelCase_     :       Any  		=					decoder_attention_heads
       UpperCAmelCase_     :       Tuple  		=					decoder_attention_heads
       UpperCAmelCase_     :       List[str]  		=					eos_token_id
       UpperCAmelCase_     :       Optional[Any]  		=					bos_token_id
       UpperCAmelCase_     :       Optional[Any]  		=					pad_token_id
       UpperCAmelCase_     :       int  		=					decoder_start_token_id
       UpperCAmelCase_     :       Optional[int]  		=					use_cache
       UpperCAmelCase_     :       List[str]  		=					max_position_embeddings

       UpperCAmelCase_     :       Tuple  		=					None
       UpperCAmelCase_     :       int  		=					decoder_seq_length
       UpperCAmelCase_     :       List[Any]  		=					2
       UpperCAmelCase_     :       str  		=					1


   def       A__    (    self:     Optional[Any]							)	->      Optional[int]:
       UpperCAmelCase_     :       Optional[int]  		=					ids_tensor([self.batch_size, self.decoder_seq_length]  ,self.vocab_size							)

       UpperCAmelCase_     :       List[str]  		=					None
       if self.use_attention_mask:
           UpperCAmelCase_     :       Optional[Any]  		=					ids_tensor([self.batch_size, self.decoder_seq_length]  ,vocab_size=2							)

       UpperCAmelCase_     :       Union[str, Any]  		=					None
       if self.use_labels:
           UpperCAmelCase_     :       List[str]  		=					ids_tensor([self.batch_size, self.decoder_seq_length]  ,self.vocab_size							)

       UpperCAmelCase_     :       Optional[int]  		=					SpeechaTextaConfig(
           vocab_size=self.vocab_size  ,d_model=self.d_model  ,decoder_layers=self.decoder_layers  ,decoder_ffn_dim=self.decoder_ffn_dim  ,decoder_attention_heads=self.decoder_attention_heads  ,eos_token_id=self.eos_token_id  ,bos_token_id=self.bos_token_id  ,use_cache=self.use_cache  ,pad_token_id=self.pad_token_id  ,decoder_start_token_id=self.decoder_start_token_id  ,max_position_embeddings=self.max_position_embeddings  ,)

       return (
           config,
           input_ids,
           attention_mask,
           lm_labels,
       )


   def       A__    (    self:     Optional[Any]  ,lowerCamelCase_:     Dict  ,lowerCamelCase_:     str  ,lowerCamelCase_:     str  ,lowerCamelCase_:     List[Any]  ,)	->      List[str]:
       UpperCAmelCase_     :       Dict  		=					True
       UpperCAmelCase_     :       Dict  		=					SpeechaTextaDecoder(config=lowerCamelCase_							).to(lowerCamelCase_							).eval()
       UpperCAmelCase_     :       Any  		=					input_ids[:2]

       input_ids[input_ids == 0] += 1
       # first forward pass
       UpperCAmelCase_     :       Optional[Any]  		=					model(lowerCamelCase_  ,use_cache=lowerCamelCase_							)
       UpperCAmelCase_     :       List[str]  		=					model(lowerCamelCase_							)
       UpperCAmelCase_     :       int  		=					model(lowerCamelCase_  ,use_cache=lowerCamelCase_							)

       self.parent.assertTrue(len(lowerCamelCase_							) == len(lowerCamelCase_							)							)
       self.parent.assertTrue(len(lowerCamelCase_							) == len(lowerCamelCase_							) + 1							)

       UpperCAmelCase_     :       Dict  		=					outputs["""past_key_values"""]

       # create hypothetical next token and extent to next_input_ids
       UpperCAmelCase_     :       Dict  		=					ids_tensor((2, 1)  ,config.vocab_size - 1							) + 1

       # append to next input_ids and
       UpperCAmelCase_     :       int  		=					torch.cat([input_ids, next_tokens]  ,dim=-1							)

       print(lowerCamelCase_							)

       UpperCAmelCase_     :       Dict  		=					model(lowerCamelCase_							)["""last_hidden_state"""]
       UpperCAmelCase_     :       int  		=					model(lowerCamelCase_  ,past_key_values=lowerCamelCase_							)["""last_hidden_state"""]

       # select random slice
       UpperCAmelCase_     :       List[str]  		=					ids_tensor((1,)  ,output_from_past.shape[-1]							).item()
       UpperCAmelCase_     :       Optional[Any]  		=					output_from_no_past[:, next_input_ids.shape[-1] - 1, random_slice_idx].detach()
       UpperCAmelCase_     :       Any  		=					output_from_past[:, 0, random_slice_idx].detach()

       # test that outputs are equal for slice
       assert torch.allclose(lowerCamelCase_  ,lowerCamelCase_  ,atol=1e-3							)






   def       A__    (    self:     Optional[Any]							)	->      List[str]:
       UpperCAmelCase_     :       Dict  		=					self.prepare_config_and_inputs()
       (
           (
           UpperCAmelCase_
       )		, (
           UpperCAmelCase_
       )		, (
           UpperCAmelCase_
       )		, (
           UpperCAmelCase_
       )		, 
       )     :       Dict  		=					config_and_inputs

       UpperCAmelCase_     :       Any  		=					{
           """input_ids""": input_ids,
           """attention_mask""": attention_mask,
       }
       return config, inputs_dict







@require_torch
class       _snake_case				(     __snake_case ,							__snake_case ,							__snake_case ,							unittest.TestCase ):






   '''simple docstring'''






   A__      : Optional[Any]       =						(SpeechaTextaDecoder, SpeechaTextaForCausalLM) if is_torch_available() else ()
   A__      : List[Any]       =						(SpeechaTextaForCausalLM,) if is_torch_available() else ()
   A__      : Any       =						{"text-generation": SpeechaTextaForCausalLM} if is_torch_available() else {}
   A__      : Optional[int]       =						True
   A__      : Union[str, Any]       =						False


   def       A__    (    self:     Tuple  ,)	->      Any:
       UpperCAmelCase_     :       Any  		=					SpeechaTextaStandaloneDecoderModelTester(self  ,is_training=lowerCamelCase_							)
       UpperCAmelCase_     :       List[Any]  		=					ConfigTester(self  ,config_class=lowerCamelCase_							)


   def       A__    (    self:     Dict							)	->      int:
       pass


   def       A__    (    self:     str							)	->      int:
       pass


   def       A__    (    self:     Union[str, Any]							)	->      Optional[int]:
       pass


   def       A__    (    self:     List[str]							)	->      int:
       self.config_tester.run_common_tests()


   def       A__    (    self:     int							)	->      Dict:
       UpperCAmelCase_     :       List[str]  		=					self.model_tester.prepare_config_and_inputs()
       self.model_tester.create_and_check_decoder_model_past(*lowerCamelCase_							)






   def       A__    (    self:     Optional[int]							)	->      Any:
       return




"""simple docstring"""







import numpy as np
import torch
import torch.nn as nn
from transformers import CLIPConfig, CLIPVisionModel, PreTrainedModel

from ...utils import logging


lowerCamelCase_  :  Tuple     		=    logging.get_logger(__name__)
def _A		(		lowercase			,       lowercase						):







    """simple docstring"""
    a											=nn.functional.normalize(lowercase						)
    a											=nn.functional.normalize(lowercase						)
    return torch.mm(lowercase			,       normalized_text_embeds.t()						)






class  __A							(			_SCREAMING_SNAKE_CASE     ):






 """simple docstring"""




 __lowerCAmelCase					   =    CLIPConfig

 __lowerCAmelCase					   =    ["CLIPEncoderLayer"]


 def __init__(						self					,       __A		)   ->	int:
     super().__init__(__A		)

     a											=CLIPVisionModel(config.vision_config		)
     a											=nn.Linear(config.vision_config.hidden_size					,       config.projection_dim					,       bias=__A		)

     a											=nn.Parameter(torch.ones(17					,       config.projection_dim		)					,       requires_grad=__A		)
     a											=nn.Parameter(torch.ones(3					,       config.projection_dim		)					,       requires_grad=__A		)

     a											=nn.Parameter(torch.ones(17		)					,       requires_grad=__A		)
     a											=nn.Parameter(torch.ones(3		)					,       requires_grad=__A		)


 @torch.no_grad()
 def  SCREAMING_SNAKE_CASE				(						self					,       __A					,       __A		)   ->	Union[str, Any]:
     a											=self.vision_model(__A		)[1]  # pooled_output
     a											=self.visual_projection(__A		)

     # we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
     a											=cosine_distance(__A					,       self.special_care_embeds		).cpu().float().numpy()
     a											=cosine_distance(__A					,       self.concept_embeds		).cpu().float().numpy()

     a											=[]
     a											=image_embeds.shape[0]
     for i in range(__A		):
         a											={'''special_scores''': {}, '''special_care''': [], '''concept_scores''': {}, '''bad_concepts''': []}

         # increase this value to create a stronger `nfsw` filter
         # at the cost of increasing the possibility of filtering benign images
         a											=0.0

         for concept_idx in range(len(special_cos_dist[0]		)		):
             a											=special_cos_dist[i][concept_idx]
             a											=self.special_care_embeds_weights[concept_idx].item()
             a											=round(concept_cos - concept_threshold + adjustment					,       3		)
             if result_img["special_scores"][concept_idx] > 0:
                 result_img["special_care"].append({concept_idx, result_img['''special_scores'''][concept_idx]}		)
                 a											=0.01

         for concept_idx in range(len(cos_dist[0]		)		):
             a											=cos_dist[i][concept_idx]
             a											=self.concept_embeds_weights[concept_idx].item()
             a											=round(concept_cos - concept_threshold + adjustment					,       3		)
             if result_img["concept_scores"][concept_idx] > 0:
                 result_img["bad_concepts"].append(__A		)

         result.append(__A		)

     a											=[len(res['''bad_concepts''']		) > 0 for res in result]

     for idx, has_nsfw_concept in enumerate(__A		):
         if has_nsfw_concept:
             if torch.is_tensor(__A		) or torch.is_tensor(images[0]		):
                 a											=torch.zeros_like(images[idx]		)  # black image
             else:
                 a											=np.zeros(images[idx].shape		)  # black image

     if any(__A		):
         logger.warning(
             '''Potential NSFW content was detected in one or more images. A black image will be returned instead.'''
             ''' Try again with a different prompt and/or seed.'''		)

     return images, has_nsfw_concepts





 @torch.no_grad()
 def  SCREAMING_SNAKE_CASE				(						self					,       __A					,       __A		)   ->	str:
     a											=self.vision_model(__A		)[1]  # pooled_output
     a											=self.visual_projection(__A		)

     a											=cosine_distance(__A					,       self.special_care_embeds		)
     a											=cosine_distance(__A					,       self.concept_embeds		)

     # increase this value to create a stronger `nsfw` filter
     # at the cost of increasing the possibility of filtering benign images
     a											=0.0

     a											=special_cos_dist - self.special_care_embeds_weights + adjustment
     # special_scores = special_scores.round(decimals=3)
     a											=torch.any(special_scores > 0					,       dim=1		)
     a											=special_care * 0.01
     a											=special_adjustment.unsqueeze(1		).expand(-1					,       cos_dist.shape[1]		)

     a											=(cos_dist - self.concept_embeds_weights) + special_adjustment
     # concept_scores = concept_scores.round(decimals=3)
     a											=torch.any(concept_scores > 0					,       dim=1		)

     a											=0.0  # black image

     return images, has_nsfw_concepts