





'''simple docstring'''




import math

import torch
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from .attention_processor import Attention
from .embeddings import get_timestep_embedding
from .modeling_utils import ModelMixin


class 					SCREAMING_SNAKE_CASE       (a__   ,						a__				):


     @register_to_config
     def __init__(		self		,      _UpperCAmelCase = 128		,      _UpperCAmelCase = 256		,      _UpperCAmelCase = 2000.0		,      _UpperCAmelCase = 768		,      _UpperCAmelCase = 12		,      _UpperCAmelCase = 12		,      _UpperCAmelCase = 64		,      _UpperCAmelCase = 2048		,      _UpperCAmelCase = 0.1		,      ):

          '''simple docstring'''






          super().__init__()

          __A		:					Tuple						=			nn.Sequential(
              nn.Linear(_UpperCAmelCase		,      d_model * 4		,      bias=_UpperCAmelCase)		,      nn.SiLU()		,      nn.Linear(d_model * 4		,      d_model * 4		,      bias=_UpperCAmelCase)		,      nn.SiLU()		,      )

          __A		:					Optional[Any]						=			nn.Embedding(_UpperCAmelCase		,      _UpperCAmelCase)
          __A		:					Dict						=			False

          __A		:					Any						=			nn.Linear(_UpperCAmelCase		,      _UpperCAmelCase		,      bias=_UpperCAmelCase)

          __A		:					Dict						=			nn.Dropout(p=_UpperCAmelCase)

          __A		:					Dict						=			nn.ModuleList()
          for lyr_num in range(_UpperCAmelCase):
               # FiLM conditional T5 decoder
               __A		:					str						=			DecoderLayer(d_model=_UpperCAmelCase		,      d_kv=_UpperCAmelCase		,      num_heads=_UpperCAmelCase		,      d_ff=_UpperCAmelCase		,      dropout_rate=_UpperCAmelCase)
               self.decoders.append(_UpperCAmelCase)

          __A		:					Dict						=			TaLayerNorm(_UpperCAmelCase)

          __A		:					Optional[int]						=			nn.Dropout(p=_UpperCAmelCase)
          __A		:					int						=			nn.Linear(_UpperCAmelCase		,      _UpperCAmelCase		,      bias=_UpperCAmelCase)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					int						=			torch.mul(query_input.unsqueeze(-1)		,      key_input.unsqueeze(-2))
          return mask.unsqueeze(-3)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          __A					,__A					,__A		:					Tuple						=			decoder_input_tokens.shape
          assert decoder_noise_time.shape == (batch,)

          # decoder_noise_time is in [0, 1), so rescale to expected timing range.
          __A		:					Union[str, Any]						=			get_timestep_embedding(
              decoder_noise_time * self.config.max_decoder_noise_time		,      embedding_dim=self.config.d_model		,      max_period=self.config.max_decoder_noise_time		,      ).to(dtype=self.dtype)

          __A		:					List[str]						=			self.conditioning_emb(_UpperCAmelCase).unsqueeze(1)

          assert conditioning_emb.shape == (batch, 1, self.config.d_model * 4)

          __A		:					Union[str, Any]						=			decoder_input_tokens.shape[1]

          # If we want to use relative positions for audio context, we can just offset
          # this sequence by the length of encodings_and_masks.
          __A		:					List[Any]						=			torch.broadcast_to(
              torch.arange(_UpperCAmelCase		,      device=decoder_input_tokens.device)		,      (batch, seq_length)		,      )

          __A		:					str						=			self.position_encoding(_UpperCAmelCase)

          __A		:					List[str]						=			self.continuous_inputs_projection(_UpperCAmelCase)
          inputs += position_encodings
          __A		:					str						=			self.dropout(_UpperCAmelCase)

          # decoder: No padding present.
          __A		:					int						=			torch.ones(
              decoder_input_tokens.shape[:2]		,      device=decoder_input_tokens.device		,      dtype=inputs.dtype)

          # Translate encoding masks to encoder-decoder masks.
          __A		:					str						=			[(x, self.encoder_decoder_mask(_UpperCAmelCase		,      _UpperCAmelCase)) for x, y in encodings_and_masks]

          # cross attend style: concat encodings
          __A		:					Union[str, Any]						=			torch.cat([x[0] for x in encodings_and_encdec_masks]		,      dim=1)
          __A		:					Dict						=			torch.cat([x[1] for x in encodings_and_encdec_masks]		,      dim=-1)

          for lyr in self.decoders:
               __A		:					List[str]						=			lyr(
                   _UpperCAmelCase		,      conditioning_emb=_UpperCAmelCase		,      encoder_hidden_states=_UpperCAmelCase		,      encoder_attention_mask=_UpperCAmelCase		,      )[0]

          __A		:					Optional[Any]						=			self.decoder_norm(_UpperCAmelCase)
          __A		:					Optional[int]						=			self.post_dropout(_UpperCAmelCase)

          __A		:					Optional[Any]						=			self.spec_out(_UpperCAmelCase)
          return spec_out




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase=1e-6):

          '''simple docstring'''






          super().__init__()
          __A		:					str						=			nn.ModuleList()

          # cond self attention: layer 0
          self.layer.append(
              TaLayerSelfAttentionCond(d_model=_UpperCAmelCase		,      d_kv=_UpperCAmelCase		,      num_heads=_UpperCAmelCase		,      dropout_rate=_UpperCAmelCase))

          # cross attention: layer 1
          self.layer.append(
              TaLayerCrossAttention(
                  d_model=_UpperCAmelCase		,      d_kv=_UpperCAmelCase		,      num_heads=_UpperCAmelCase		,      dropout_rate=_UpperCAmelCase		,      layer_norm_epsilon=_UpperCAmelCase		,      ))

          # Film Cond MLP + dropout: last layer
          self.layer.append(
              TaLayerFFCond(d_model=_UpperCAmelCase		,      d_ff=_UpperCAmelCase		,      dropout_rate=_UpperCAmelCase		,      layer_norm_epsilon=_UpperCAmelCase))


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase=None		,      _UpperCAmelCase=None		,      _UpperCAmelCase=None		,      _UpperCAmelCase=None		,      _UpperCAmelCase=None		,      ):

          '''simple docstring'''






          __A		:					Union[str, Any]						=			self.layer[0](
              _UpperCAmelCase		,      conditioning_emb=_UpperCAmelCase		,      attention_mask=_UpperCAmelCase		,      )

          if encoder_hidden_states is not None:
               __A		:					Optional[Any]						=			torch.where(encoder_attention_mask > 0		,      0		,      -1e1_0).to(
                   encoder_hidden_states.dtype)

               __A		:					Dict						=			self.layer[1](
                   _UpperCAmelCase		,      key_value_states=_UpperCAmelCase		,      attention_mask=_UpperCAmelCase		,      )

          # Apply Film Conditional Feed Forward layer
          __A		:					List[Any]						=			self.layer[-1](_UpperCAmelCase		,      _UpperCAmelCase)

          return (hidden_states,)




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()
          __A		:					int						=			TaLayerNorm(_UpperCAmelCase)
          __A		:					int						=			TaFiLMLayer(in_features=d_model * 4		,      out_features=_UpperCAmelCase)
          __A		:					Optional[Any]						=			Attention(query_dim=_UpperCAmelCase		,      heads=_UpperCAmelCase		,      dim_head=_UpperCAmelCase		,      out_bias=_UpperCAmelCase		,      scale_qk=_UpperCAmelCase)
          __A		:					Dict						=			nn.Dropout(_UpperCAmelCase)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase=None		,      _UpperCAmelCase=None		,      ):

          '''simple docstring'''






          __A		:					List[str]						=			self.layer_norm(_UpperCAmelCase)

          if conditioning_emb is not None:
               __A		:					List[Any]						=			self.FiLMLayer(_UpperCAmelCase		,      _UpperCAmelCase)

          # Self-attention block
          __A		:					Any						=			self.attention(_UpperCAmelCase)

          __A		:					Any						=			hidden_states + self.dropout(_UpperCAmelCase)

          return hidden_states




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()
          __A		:					Any						=			Attention(query_dim=_UpperCAmelCase		,      heads=_UpperCAmelCase		,      dim_head=_UpperCAmelCase		,      out_bias=_UpperCAmelCase		,      scale_qk=_UpperCAmelCase)
          __A		:					List[str]						=			TaLayerNorm(_UpperCAmelCase		,      eps=_UpperCAmelCase)
          __A		:					List[Any]						=			nn.Dropout(_UpperCAmelCase)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase=None		,      _UpperCAmelCase=None		,      ):

          '''simple docstring'''






          __A		:					Tuple						=			self.layer_norm(_UpperCAmelCase)
          __A		:					str						=			self.attention(
              _UpperCAmelCase		,      encoder_hidden_states=_UpperCAmelCase		,      attention_mask=attention_mask.squeeze(1)		,      )
          __A		:					List[Any]						=			hidden_states + self.dropout(_UpperCAmelCase)
          return layer_output




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()
          __A		:					Any						=			TaDenseGatedActDense(d_model=_UpperCAmelCase		,      d_ff=_UpperCAmelCase		,      dropout_rate=_UpperCAmelCase)
          __A		:					List[Any]						=			TaFiLMLayer(in_features=d_model * 4		,      out_features=_UpperCAmelCase)
          __A		:					int						=			TaLayerNorm(_UpperCAmelCase		,      eps=_UpperCAmelCase)
          __A		:					Optional[int]						=			nn.Dropout(_UpperCAmelCase)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase=None):

          '''simple docstring'''






          __A		:					Optional[Any]						=			self.layer_norm(_UpperCAmelCase)
          if conditioning_emb is not None:
               __A		:					Optional[int]						=			self.film(_UpperCAmelCase		,      _UpperCAmelCase)

          __A		:					Optional[int]						=			self.DenseReluDense(_UpperCAmelCase)
          __A		:					Any						=			hidden_states + self.dropout(_UpperCAmelCase)
          return hidden_states




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()
          __A		:					List[str]						=			nn.Linear(_UpperCAmelCase		,      _UpperCAmelCase		,      bias=_UpperCAmelCase)
          __A		:					List[Any]						=			nn.Linear(_UpperCAmelCase		,      _UpperCAmelCase		,      bias=_UpperCAmelCase)
          __A		:					Optional[int]						=			nn.Linear(_UpperCAmelCase		,      _UpperCAmelCase		,      bias=_UpperCAmelCase)
          __A		:					Tuple						=			nn.Dropout(_UpperCAmelCase)
          __A		:					Tuple						=			NewGELUActivation()


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					str						=			self.act(self.wi_a(_UpperCAmelCase))
          __A		:					List[Any]						=			self.wi_a(_UpperCAmelCase)
          __A		:					int						=			hidden_gelu * hidden_linear
          __A		:					List[Any]						=			self.dropout(_UpperCAmelCase)

          __A		:					int						=			self.wo(_UpperCAmelCase)
          return hidden_states




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase=1e-6):

          '''simple docstring'''






          super().__init__()
          __A		:					Optional[int]						=			nn.Parameter(torch.ones(_UpperCAmelCase))
          __A		:					Optional[Any]						=			eps


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					str						=			hidden_states.to(torch.floataa).pow(2).mean(-1		,      keepdim=_UpperCAmelCase)
          __A		:					str						=			hidden_states * torch.rsqrt(variance + self.variance_epsilon)

          # convert into half-precision if necessary
          if self.weight.dtype in [torch.floataa, torch.bfloataa]:
               __A		:					Optional[Any]						=			hidden_states.to(self.weight.dtype)

          return self.weight * hidden_states




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase):

          '''simple docstring'''






          return 0.5 * input * (1.0 + torch.tanh(math.sqrt(2.0 / math.pi) * (input + 0.044715 * torch.pow(_UpperCAmelCase		,      3.0))))




class 					SCREAMING_SNAKE_CASE       (nn.Module				):


     def __init__(		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          super().__init__()
          __A		:					Tuple						=			nn.Linear(_UpperCAmelCase		,      out_features * 2		,      bias=_UpperCAmelCase)


     def 				SCREAMING_SNAKE_CASE    (		self		,      _UpperCAmelCase		,      _UpperCAmelCase):

          '''simple docstring'''






          __A		:					Dict						=			self.scale_bias(_UpperCAmelCase)
          __A					,__A		:					str						=			torch.chunk(_UpperCAmelCase		,      2		,      -1)
          __A		:					Optional[Any]						=			x * (1 + scale) + shift
          return x

from typing import Any, Dict, List, Union

from ..utils import add_end_docstrings, is_torch_available, is_vision_available, logging, requires_backends
from .base import PIPELINE_INIT_ARGS, ChunkPipeline


if is_vision_available():
     from PIL import Image

     from ..image_utils import load_image

if is_torch_available():
     import torch

     from transformers.modeling_outputs import BaseModelOutput

     from ..models.auto.modeling_auto import MODEL_FOR_ZERO_SHOT_OBJECT_DETECTION_MAPPING

lowerCamelCase__     :		Optional[Any]        = logging.get_logger(__name__)



@add_end_docstrings(UpperCAmelCase_				)
class    _snake_case (						UpperCAmelCase_				):


 def __init__(       self       ,	**SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   super().__init__(**SCREAMING_SNAKE_CASE_)

   if self.framework == "tf":
     raise ValueError(f'The {self.__class__} is only available in PyTorch.')

   requires_backends(self       ,	"""vision""")
   self.check_model_type(SCREAMING_SNAKE_CASE_)


 def __call__(       self       ,	SCREAMING_SNAKE_CASE_       ,	SCREAMING_SNAKE_CASE_ = None       ,	**SCREAMING_SNAKE_CASE_       ,	):


   '''simple docstring'''




   if "text_queries" in kwargs:
     lowercase__	: Any     		=   kwargs.pop("""text_queries""")

   if isinstance(SCREAMING_SNAKE_CASE_       ,	(str, Image.Image)):
     lowercase__	: Optional[Any]     		=   {"""image""": image, """candidate_labels""": candidate_labels}
   else:
     lowercase__	: int     		=   image
   lowercase__	: List[str]     		=   super().__call__(SCREAMING_SNAKE_CASE_       ,	**SCREAMING_SNAKE_CASE_)
   return results


 def   lowercase__  (       self       ,	**SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   lowercase__	: Tuple     		=   {}
   if "threshold" in kwargs:
     lowercase__	: List[Any]     		=   kwargs["""threshold"""]
   if "top_k" in kwargs:
     lowercase__	: int     		=   kwargs["""top_k"""]
   return {}, {}, postprocess_params


 def   lowercase__  (       self       ,	SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   lowercase__	: str     		=   load_image(inputs["""image"""])
   lowercase__	: Any     		=   inputs["""candidate_labels"""]
   if isinstance(SCREAMING_SNAKE_CASE_       ,	SCREAMING_SNAKE_CASE_):
     lowercase__	: List[str]     		=   candidate_labels.split(""",""")

   lowercase__	: Tuple     		=   torch.tensor([[image.height, image.width]]       ,	dtype=torch.intaa)
   for i, candidate_label in enumerate(SCREAMING_SNAKE_CASE_):
     lowercase__	: Optional[Any]     		=   self.tokenizer(SCREAMING_SNAKE_CASE_       ,	return_tensors=self.framework)
     lowercase__	: Union[str, Any]     		=   self.image_processor(SCREAMING_SNAKE_CASE_       ,	return_tensors=self.framework)
     yield {
         "is_last": i == len(SCREAMING_SNAKE_CASE_) - 1,
         "target_size": target_size,
         "candidate_label": candidate_label,
         **text_inputs,
         **image_features,
     }


 def   lowercase__  (       self       ,	SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   lowercase__	: str     		=   model_inputs.pop("""target_size""")
   lowercase__	: Optional[int]     		=   model_inputs.pop("""candidate_label""")
   lowercase__	: Dict     		=   model_inputs.pop("""is_last""")

   lowercase__	: Union[str, Any]     		=   self.model(**SCREAMING_SNAKE_CASE_)

   lowercase__	: Union[str, Any]     		=   {"""target_size""": target_size, """candidate_label""": candidate_label, """is_last""": is_last, **outputs}
   return model_outputs


 def   lowercase__  (       self       ,	SCREAMING_SNAKE_CASE_       ,	SCREAMING_SNAKE_CASE_=0.1       ,	SCREAMING_SNAKE_CASE_=None):


   '''simple docstring'''




   lowercase__	: Union[str, Any]     		=   []
   for model_output in model_outputs:
     lowercase__	: Optional[int]     		=   model_output["""candidate_label"""]
     lowercase__	: Tuple     		=   BaseModelOutput(SCREAMING_SNAKE_CASE_)
     lowercase__	: List[str]     		=   self.image_processor.post_process_object_detection(
         outputs=SCREAMING_SNAKE_CASE_       ,	threshold=SCREAMING_SNAKE_CASE_       ,	target_sizes=model_output["""target_size"""])[0]

     for index in outputs["scores"].nonzero():
       lowercase__	: Optional[Any]     		=   outputs["""scores"""][index].item()
       lowercase__	: Optional[Any]     		=   self._get_bounding_box(outputs["""boxes"""][index][0])

       lowercase__	: Tuple     		=   {"""score""": score, """label""": label, """box""": box}
       results.append(SCREAMING_SNAKE_CASE_)

   lowercase__	: int     		=   sorted(SCREAMING_SNAKE_CASE_       ,	key=lambda SCREAMING_SNAKE_CASE_: x["score"]       ,	reverse=SCREAMING_SNAKE_CASE_)
   if top_k:
     lowercase__	: Any     		=   results[:top_k]

   return results



 def   lowercase__  (       self       ,	SCREAMING_SNAKE_CASE_):


   '''simple docstring'''




   if self.framework != "pt":
     raise ValueError("""The ZeroShotObjectDetectionPipeline is only available in PyTorch.""")
   lowercase__		,     lowercase__		,     lowercase__		,     lowercase__	: List[Any]     		=   box.int().tolist()
   lowercase__	: Optional[int]     		=   {
       """xmin""": xmin,
       """ymin""": ymin,
       """xmax""": xmax,
       """ymax""": ymax,
   }
   return bbox






