

from dataclasses import dataclass
from typing import Dict, Optional, Union

import torch
import torch.nn.functional as F
from torch import nn

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput
from .attention import BasicTransformerBlock
from .attention_processor import AttentionProcessor, AttnProcessor
from .embeddings import TimestepEmbedding, Timesteps
from .modeling_utils import ModelMixin




@dataclass
class      UpperCamelCase (     SCREAMING_SNAKE_CASE   ):
    __UpperCamelCase         =42




class      UpperCamelCase (     SCREAMING_SNAKE_CASE	,			SCREAMING_SNAKE_CASE   ):

    @register_to_config
    def __init__(			self :      List[str]							,    snake_case__ :      int = 3_2							,    snake_case__ :      int = 6_4							,    snake_case__ :      int = 2_0							,    snake_case__ :      int = 7_6_8							,    snake_case__ :      Union[str, Any]=7_7							,    snake_case__ :      str=4							,    snake_case__ :      float = 0.0							,    snake_case__ :      str = "silu"							,    snake_case__ :      Optional[str] = None							,    snake_case__ :      Optional[str] = None							,    snake_case__ :      Optional[str] = "linear"							,    snake_case__ :      Optional[str] = "prd"							,    snake_case__ :      Optional[int] = None							,    snake_case__ :      Optional[int] = None							,    snake_case__ :      Optional[int] = None							,    ):

     """simple docstring"""






     super().__init__()
     SCREAMING_SNAKE_CASE								=			num_attention_heads
     SCREAMING_SNAKE_CASE								=			attention_head_dim
     SCREAMING_SNAKE_CASE								=			num_attention_heads * attention_head_dim
     SCREAMING_SNAKE_CASE								=			additional_embeddings

     SCREAMING_SNAKE_CASE								=			time_embed_dim or inner_dim
     SCREAMING_SNAKE_CASE								=			embedding_proj_dim or embedding_dim
     SCREAMING_SNAKE_CASE								=			clip_embed_dim or embedding_dim

     SCREAMING_SNAKE_CASE								=			Timesteps(snake_case__							,    snake_case__							,    0    )
     SCREAMING_SNAKE_CASE								=			TimestepEmbedding(snake_case__							,    snake_case__							,    out_dim=snake_case__							,    act_fn=snake_case__    )

     SCREAMING_SNAKE_CASE								=			nn.Linear(snake_case__							,    snake_case__    )

     if embedding_proj_norm_type is None:
      SCREAMING_SNAKE_CASE								=			None
     elif embedding_proj_norm_type == "layer":
      SCREAMING_SNAKE_CASE								=			nn.LayerNorm(snake_case__    )
     else:
      raise ValueError(F"""unsupported embedding_proj_norm_type: {embedding_proj_norm_type}"""    )

     SCREAMING_SNAKE_CASE								=			nn.Linear(snake_case__							,    snake_case__    )

     if encoder_hid_proj_type is None:
      SCREAMING_SNAKE_CASE								=			None
     elif encoder_hid_proj_type == "linear":
      SCREAMING_SNAKE_CASE								=			nn.Linear(snake_case__							,    snake_case__    )
     else:
      raise ValueError(F"""unsupported encoder_hid_proj_type: {encoder_hid_proj_type}"""    )

     SCREAMING_SNAKE_CASE								=			nn.Parameter(torch.zeros(1							,    num_embeddings + additional_embeddings							,    snake_case__    )    )

     if added_emb_type == "prd":
      SCREAMING_SNAKE_CASE								=			nn.Parameter(torch.zeros(1							,    1							,    snake_case__    )    )
     elif added_emb_type is None:
      SCREAMING_SNAKE_CASE								=			None
     else:
      raise ValueError(
          F"""`added_emb_type`: {added_emb_type} is not supported. Make sure to choose one of `'prd'` or `None`."""    )

     SCREAMING_SNAKE_CASE								=			nn.ModuleList(
         [
             BasicTransformerBlock(
                 snake_case__							,    snake_case__							,    snake_case__							,    dropout=snake_case__							,    activation_fn='gelu'							,    attention_bias=snake_case__							,    )
             for d in range(snake_case__    )
         ]    )

     if norm_in_type == "layer":
      SCREAMING_SNAKE_CASE								=			nn.LayerNorm(snake_case__    )
     elif norm_in_type is None:
      SCREAMING_SNAKE_CASE								=			None
     else:
      raise ValueError(F"""Unsupported norm_in_type: {norm_in_type}."""    )

     SCREAMING_SNAKE_CASE								=			nn.LayerNorm(snake_case__    )

     SCREAMING_SNAKE_CASE								=			nn.Linear(snake_case__							,    snake_case__    )

     SCREAMING_SNAKE_CASE								=			torch.full(
         [num_embeddings + additional_embeddings, num_embeddings + additional_embeddings]							,    -10_000.0    )
     causal_attention_mask.triu_(1    )
     SCREAMING_SNAKE_CASE								=			causal_attention_mask[None, ...]
     self.register_buffer('causal_attention_mask'							,    snake_case__							,    persistent=snake_case__    )

     SCREAMING_SNAKE_CASE								=			nn.Parameter(torch.zeros(1							,    snake_case__    )    )
     SCREAMING_SNAKE_CASE								=			nn.Parameter(torch.zeros(1							,    snake_case__    )    )

    @property
    # Copied from diffusers.models.unet_2d_condition.UNet2DConditionModel.attn_processors
    def UpperCamelCase			(			self :      Optional[Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			{}

     def fn_recursive_add_processors(snake_case__ :      str							,    snake_case__ :      torch.nn.Module							,    snake_case__ :      Dict[str, AttentionProcessor]    ):
      if hasattr(snake_case__							,    'set_processor'    ):
       SCREAMING_SNAKE_CASE								=			module.processor

      for sub_name, child in module.named_children():
       fn_recursive_add_processors(F"""{name}.{sub_name}"""							,    snake_case__							,    snake_case__    )

      return processors

     for name, module in self.named_children():
      fn_recursive_add_processors(snake_case__							,    snake_case__							,    snake_case__    )

     return processors

    def UpperCamelCase			(			self :      Dict							,    snake_case__ :      Union[AttentionProcessor, Dict[str, AttentionProcessor]]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			len(self.attn_processors.keys()    )

     if isinstance(snake_case__							,    snake_case__    ) and len(snake_case__    ) != count:
      raise ValueError(
          F"""A dict of processors was passed, but the number of processors {len(snake_case__    )} does not match the"""
          F""" number of attention layers: {count}. Please make sure to pass {count} processor classes."""    )

     def fn_recursive_attn_processor(snake_case__ :      str							,    snake_case__ :      torch.nn.Module							,    snake_case__ :      List[str]    ):
      if hasattr(snake_case__							,    'set_processor'    ):
       if not isinstance(snake_case__							,    snake_case__    ):
        module.set_processor(snake_case__    )
       else:
        module.set_processor(processor.pop(F"""{name}.processor"""    )    )

      for sub_name, child in module.named_children():
       fn_recursive_attn_processor(F"""{name}.{sub_name}"""							,    snake_case__							,    snake_case__    )

     for name, module in self.named_children():
      fn_recursive_attn_processor(snake_case__							,    snake_case__							,    snake_case__    )

    def UpperCamelCase			(			self :      Optional[Any]    ):

     """simple docstring"""






     self.set_attn_processor(AttnProcessor()    )

    def UpperCamelCase			(			self :      Any							,    snake_case__ :      str							,    snake_case__ :      Union[torch.Tensor, float, int]							,    snake_case__ :      torch.FloatTensor							,    snake_case__ :      Optional[torch.FloatTensor] = None							,    snake_case__ :      Optional[torch.BoolTensor] = None							,    snake_case__ :      bool = True							,    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			hidden_states.shape[0]

     SCREAMING_SNAKE_CASE								=			timestep
     if not torch.is_tensor(snake_case__    ):
      SCREAMING_SNAKE_CASE								=			torch.tensor([timesteps]							,    dtype=torch.long							,    device=hidden_states.device    )
     elif torch.is_tensor(snake_case__    ) and len(timesteps.shape    ) == 0:
      SCREAMING_SNAKE_CASE								=			timesteps[None].to(hidden_states.device    )

     # broadcast to batch dimension in a way that's compatible with ONNX/Core ML
     SCREAMING_SNAKE_CASE								=			timesteps * torch.ones(snake_case__							,    dtype=timesteps.dtype							,    device=timesteps.device    )

     SCREAMING_SNAKE_CASE								=			self.time_proj(snake_case__    )

     # timesteps does not contain any weights and will always return f32 tensors
     # but time_embedding might be fp16, so we need to cast here.
     SCREAMING_SNAKE_CASE								=			timesteps_projected.to(dtype=self.dtype    )
     SCREAMING_SNAKE_CASE								=			self.time_embedding(snake_case__    )

     if self.embedding_proj_norm is not None:
      SCREAMING_SNAKE_CASE								=			self.embedding_proj_norm(snake_case__    )

     SCREAMING_SNAKE_CASE								=			self.embedding_proj(snake_case__    )
     if self.encoder_hidden_states_proj is not None and encoder_hidden_states is not None:
      SCREAMING_SNAKE_CASE								=			self.encoder_hidden_states_proj(snake_case__    )
     elif self.encoder_hidden_states_proj is not None and encoder_hidden_states is None:
      raise ValueError('`encoder_hidden_states_proj` requires `encoder_hidden_states` to be set'    )

     SCREAMING_SNAKE_CASE								=			self.proj_in(snake_case__    )

     SCREAMING_SNAKE_CASE								=			self.positional_embedding.to(hidden_states.dtype    )

     SCREAMING_SNAKE_CASE								=			[]
     SCREAMING_SNAKE_CASE								=			0

     if encoder_hidden_states is not None:
      additional_embeds.append(snake_case__    )
      additional_embeddings_len += encoder_hidden_states.shape[1]

     if len(proj_embeddings.shape    ) == 2:
      SCREAMING_SNAKE_CASE								=			proj_embeddings[:, None, :]

     if len(hidden_states.shape    ) == 2:
      SCREAMING_SNAKE_CASE								=			hidden_states[:, None, :]

     SCREAMING_SNAKE_CASE								=			additional_embeds + [
         proj_embeddings,
         time_embeddings[:, None, :],
         hidden_states,
     ]

     if self.prd_embedding is not None:
      SCREAMING_SNAKE_CASE								=			self.prd_embedding.to(hidden_states.dtype    ).expand(snake_case__							,    -1							,    -1    )
      additional_embeds.append(snake_case__    )

     SCREAMING_SNAKE_CASE								=			torch.cat(
         snake_case__							,    dim=1							,    )

     # Allow positional_embedding to not include the `addtional_embeddings` and instead pad it with zeros for these additional tokens
     SCREAMING_SNAKE_CASE								=			additional_embeddings_len + proj_embeddings.shape[1] + 1
     if positional_embeddings.shape[1] < hidden_states.shape[1]:
      SCREAMING_SNAKE_CASE								=			F.pad(
          snake_case__							,    (
              0,
              0,
              additional_embeddings_len,
              self.prd_embedding.shape[1] if self.prd_embedding is not None else 0,
          )							,    value=0.0							,    )

     SCREAMING_SNAKE_CASE								=			hidden_states + positional_embeddings

     if attention_mask is not None:
      SCREAMING_SNAKE_CASE								=			(1 - attention_mask.to(hidden_states.dtype    )) * -10_000.0
      SCREAMING_SNAKE_CASE								=			F.pad(snake_case__							,    (0, self.additional_embeddings)							,    value=0.0    )
      SCREAMING_SNAKE_CASE								=			(attention_mask[:, None, :] + self.causal_attention_mask).to(hidden_states.dtype    )
      SCREAMING_SNAKE_CASE								=			attention_mask.repeat_interleave(self.config.num_attention_heads							,    dim=0    )

     if self.norm_in is not None:
      SCREAMING_SNAKE_CASE								=			self.norm_in(snake_case__    )

     for block in self.transformer_blocks:
      SCREAMING_SNAKE_CASE								=			block(snake_case__							,    attention_mask=snake_case__    )

     SCREAMING_SNAKE_CASE								=			self.norm_out(snake_case__    )

     if self.prd_embedding is not None:
      SCREAMING_SNAKE_CASE								=			hidden_states[:, -1]
     else:
      SCREAMING_SNAKE_CASE								=			hidden_states[:, additional_embeddings_len:]

     SCREAMING_SNAKE_CASE								=			self.proj_to_clip_embeddings(snake_case__    )

     if not return_dict:
      return (predicted_image_embedding,)

     return PriorTransformerOutput(predicted_image_embedding=snake_case__    )




    def UpperCamelCase			(			self :      List[str]							,    snake_case__ :      int    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			(prior_latents * self.clip_std) + self.clip_mean
     return prior_latents




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

a_    :						Dict    		=      logging.get_logger(__name__)




@add_end_docstrings(SCREAMING_SNAKE_CASE   )
class      UpperCamelCase (     SCREAMING_SNAKE_CASE   ):

    def __init__(			self :      Any							,    **snake_case__ :      Optional[int]    ):

     """simple docstring"""






     super().__init__(**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(snake_case__    )

    def __call__(			self :      List[Any]							,    snake_case__ :      Union[str, "Image.Image", List[Dict[str, Any]]]							,    snake_case__ :      Union[str, List[str]] = None							,    **snake_case__ :      Union[str, Any]							,    ):

     """simple docstring"""






     if "text_queries" in kwargs:
      SCREAMING_SNAKE_CASE								=			kwargs.pop('text_queries'    )

     if isinstance(snake_case__							,    (str, Image.Image)    ):
      SCREAMING_SNAKE_CASE								=			{'image': image, 'candidate_labels': candidate_labels}
     else:
      SCREAMING_SNAKE_CASE								=			image
     SCREAMING_SNAKE_CASE								=			super().__call__(snake_case__							,    **snake_case__    )
     return results

    def UpperCamelCase			(			self :      Union[str, Any]							,    **snake_case__ :      Optional[Any]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			{}
     if "threshold" in kwargs:
      SCREAMING_SNAKE_CASE								=			kwargs['threshold']
     if "top_k" in kwargs:
      SCREAMING_SNAKE_CASE								=			kwargs['top_k']
     return {}, {}, postprocess_params

    def UpperCamelCase			(			self :      List[Any]							,    snake_case__ :      Optional[int]    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			load_image(inputs['image']    )
     SCREAMING_SNAKE_CASE								=			inputs['candidate_labels']
     if isinstance(snake_case__							,    snake_case__    ):
      SCREAMING_SNAKE_CASE								=			candidate_labels.split(','    )

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

    def UpperCamelCase			(			self :      Any							,    snake_case__ :      Dict    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			model_inputs.pop('target_size'    )
     SCREAMING_SNAKE_CASE								=			model_inputs.pop('candidate_label'    )
     SCREAMING_SNAKE_CASE								=			model_inputs.pop('is_last'    )

     SCREAMING_SNAKE_CASE								=			self.model(**snake_case__    )

     SCREAMING_SNAKE_CASE								=			{'target_size': target_size, 'candidate_label': candidate_label, 'is_last': is_last, **outputs}
     return model_outputs

    def UpperCamelCase			(			self :      Union[str, Any]							,    snake_case__ :      Tuple							,    snake_case__ :      str=0.1							,    snake_case__ :      Union[str, Any]=None    ):

     """simple docstring"""






     SCREAMING_SNAKE_CASE								=			[]
     for model_output in model_outputs:
      SCREAMING_SNAKE_CASE								=			model_output['candidate_label']
      SCREAMING_SNAKE_CASE								=			BaseModelOutput(snake_case__    )
      SCREAMING_SNAKE_CASE								=			self.image_processor.post_process_object_detection(
          outputs=snake_case__							,    threshold=snake_case__							,    target_sizes=model_output['target_size']    )[0]

      for index in outputs["scores"].nonzero():
       SCREAMING_SNAKE_CASE								=			outputs['scores'][index].item()
       SCREAMING_SNAKE_CASE								=			self._get_bounding_box(outputs['boxes'][index][0]    )

       SCREAMING_SNAKE_CASE								=			{'score': score, 'label': label, 'box': box}
       results.append(snake_case__    )

     SCREAMING_SNAKE_CASE								=			sorted(snake_case__							,    key=lambda snake_case__    : x["score"]							,    reverse=snake_case__    )
     if top_k:
      SCREAMING_SNAKE_CASE								=			results[:top_k]

     return results




    def UpperCamelCase			(			self :      List[Any]							,    snake_case__ :      "torch.Tensor"    ):

     """simple docstring"""






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

